mesh-host/internal/selfcompose/composedstate.go

package selfcompose

import (
	"bytes"
	"context"
	"fmt"
	"maps"
	"sort"
	"strings"
	"time"

	"git.novox.be/novox/mesh-sdk/go/catalogue"
	"git.novox.be/novox/mesh-sdk/go/delivery"

	"github.com/novox/mesh-host/internal/agent"
	"github.com/novox/mesh-host/internal/compose/frombus"
)

// **The compositions this node makes, published in its composition state** (novox/hq ADR 0593 §6). The node's engine
// alone composes, and it writes every composition it makes into `composed-<node>` (delivery.ComposedBucketFor), a
// record for the provisioner, the assigner and the view, never an instruction: the engine applies only what it composes
// itself, never a value read from the state, so a value carries no signature. After a pass that composed, it writes:
//
//   - each module at the fingerprint its want names, as the agent applies it (`<node>.<module>.<fp>`), and the node's own
//     composition (`<node>.node`) when its want exists;
//   - each module placed on the node at its newest done build: the indexer's fingerprint once the builder holds a done
//     build of it, so the plan exists before the want moves (build done, the node composes and writes, the provisioner
//     reads, the providers provide, the assigner moves the want); a module it runs that no want names, at what runs;
//   - each module's plan (catalogue.Plan) exactly when the provisioner's plans name it, from the same pass's builds;
//   - a module refused alone at a fingerprint as that key with its refusal (Refused), and no resources.
//
// A key is written only when its value differs from what the state holds (read once per run), and a module's key is
// deleted once its fingerprint is not among the five it wrote last (delivery.ComposedKeep) and no want names it. A pass
// refused whole, or a bus that did not answer, writes nothing: what was written stands, and each module's have says why.

// ComposedStore is this node's composition state, bound and never created (the grants part creates it).
type ComposedStore interface {
	// Entries is every key the state holds with its value and revision.
	Entries(ctx context.Context) (map[string]ComposedEntry, error)
	Put(ctx context.Context, key string, value []byte) (revision uint64, err error)
	Delete(ctx context.Context, key string) error
}

// ComposedEntry is one key's value and revision in the state.
type ComposedEntry struct {
	Value    []byte
	Revision uint64
}

// published is what a pass composed at the wanted builds, for publishComposed: under mu.
type published struct {
	scope frombus.Scope
	fps   map[string]string // by module, the wanted fingerprint
	comps map[string]delivery.Composed
	held  map[string]string
	// placed is the modules the node's placement puts here, ran what runs of each module (by its composition's FP).
	placed []string
	ran    map[string]string
}

// publishComposed writes the node's composition state from the last pass that composed, when one asked it (pubDue).
func (c *Composer) publishComposed(ctx context.Context) {
	c.mu.Lock()
	due, at := c.pubDue, c.pubAt
	c.pubDue = false
	c.mu.Unlock()
	if !due || c.env.ComposedState == nil || at == nil {
		return
	}
	wrote, deleted, err := c.writeComposed(ctx, at)
	if err != nil {
		c.sayComposed("this node's compositions are not kept in its composition state; " + c.oweComposed() + ": " + err.Error())
		return
	}
	c.mu.Lock()
	c.pubTries = 0
	c.mu.Unlock()
	c.sayComposed("")
	if wrote+deleted > 0 {
		c.env.Say(fmt.Sprintf("this node's composition state: %d key(s) written, %d deleted", wrote, deleted))
	}
}

// newestFPs is, of the modules placed here, the fingerprint each is composed at for the state beyond the wanted ones:
// the indexer's, where the builder holds a done build of it; else the want's; else what runs of it. A module with none
// is left out.
func (c *Composer) newestFPs(ctx context.Context, at *published) map[string]string {
	out := map[string]string{}
	r := reader{c: c}
	for _, module := range at.placed {
		c.mu.Lock()
		fp2 := c.modules[module].FP2
		c.mu.Unlock()
		if fp2 != "" {
			if b, err := r.Build(ctx, module, fp2); err == nil && b != nil && b.State == delivery.BuildDone {
				out[module] = fp2
				continue
			}
		}
		if fp, ok := at.fps[module]; ok {
			out[module] = fp
		} else if fp, ok := at.ran[module]; ok {
			out[module] = fp
		}
	}
	return out
}

// writeComposed works out the state's keys from at and writes those that differ, then deletes what is past keeping.
func (c *Composer) writeComposed(ctx context.Context, at *published) (wrote, deleted int, err error) {
	values, err := c.composedValues(ctx, at)
	if err != nil {
		return 0, 0, err
	}
	c.pubWrite.Lock()
	defer c.pubWrite.Unlock()
	if c.pubKept == nil {
		kept, err := c.env.ComposedState.Entries(ctx)
		if err != nil {
			return 0, 0, fmt.Errorf("the state cannot be read: %w", err)
		}
		c.pubKept = kept
	}
	keys := make([]string, 0, len(values))
	for k := range values {
		keys = append(keys, k)
	}
	sort.Strings(keys)
	for _, key := range keys {
		if bytes.Equal(c.pubKept[key].Value, values[key]) {
			continue
		}
		rev, err := c.env.ComposedState.Put(ctx, key, values[key])
		if err != nil {
			return wrote, deleted, fmt.Errorf("%s cannot be written: %w", key, err)
		}
		c.pubKept[key] = ComposedEntry{Value: values[key], Revision: rev}
		wrote++
	}
	for _, key := range c.pastKeeping() {
		if err := c.env.ComposedState.Delete(ctx, key); err != nil {
			return wrote, deleted, fmt.Errorf("%s cannot be deleted: %w", key, err)
		}
		delete(c.pubKept, key)
		deleted++
	}
	return wrote, deleted, nil
}

// composedValues is every key the state should hold after the pass at, encoded.
func (c *Composer) composedValues(ctx context.Context, at *published) (map[string][]byte, error) {
	r := reader{c: c}
	out := map[string][]byte{}
	put := func(comp delivery.Composed, plans map[string]catalogue.Plan) error {
		if comp.Module != "" && comp.Refused == "" {
			if p, ok := plans[comp.Module]; ok {
				p := p
				comp.Plan = &p
			}
		}
		key, err := composedKeyOf(comp)
		if err != nil {
			return err
		}
		raw, err := delivery.EncodeComposed(comp)
		if err != nil {
			return fmt.Errorf("the composition %s cannot be written: %w", key, err)
		}
		out[key] = raw
		return nil
	}
	runs := []struct {
		scope frombus.Scope
		fps   map[string]string
		comps map[string]delivery.Composed
		held  map[string]string
	}{{at.scope, at.fps, at.comps, at.held}}
	// The newest builds, composed only where they differ from the wanted ones.
	newest := c.newestFPs(ctx, at)
	if !maps.Equal(newest, at.fps) {
		scope := at.scope
		scope.KeptAt = map[string]string{}
		var inOwn []string
		c.mu.Lock()
		for module, fp := range newest {
			if fp != c.modules[module].FP2 {
				scope.KeptAt[module] = fp
			}
		}
		c.mu.Unlock()
		for _, m := range at.scope.InOwn {
			if _, given := newest[m]; !given {
				inOwn = append(inOwn, m)
			}
		}
		scope.InOwn = inOwn
		comps, _, held, err := frombus.Compose(ctx, r, c.env.Mesh, scope, newest)
		switch {
		case err != nil && BusFailed(err):
			return nil, fmt.Errorf("the newest builds could not be composed: %w", err)
		case err != nil:
			// A refusal of the node at its newest builds is no refusal of what it applies: the wanted keys are written,
			// and the newest are said (each module's have says what holds it when its want moves).
			c.sayComposed("this node's newest builds do not compose, so its composition state holds the wanted ones alone: " + err.Error())
		default:
			for key, comp := range comps {
				comps[key] = c.keepRunning(comp, scope.Running)
			}
			runs = append(runs, struct {
				scope frombus.Scope
				fps   map[string]string
				comps map[string]delivery.Composed
				held  map[string]string
			}{scope, newest, comps, held})
		}
	}
	for i, run := range runs {
		plans, _, err := frombus.PlansHeld(ctx, r, c.env.Mesh, run.scope)
		if err != nil && BusFailed(err) {
			return nil, fmt.Errorf("the plans could not be composed: %w", err)
		}
		// A node whose plans refuse it whole hands the provisioner nothing, as the provisioner's own composing did: no
		// key carries a plan then.
		for _, comp := range run.comps {
			if comp.Module == "" && i > 0 {
				continue // the node's own composition is the wanted pass's alone
			}
			if err := put(comp, plans); err != nil {
				return nil, err
			}
		}
		for module, why := range run.held {
			fp, ok := run.fps[module]
			if !ok || module == agent.NodeModule {
				continue
			}
			if err := put(delivery.Composed{Node: c.env.Node, Module: module, FP: fp, Refused: why}, nil); err != nil {
				return nil, err
			}
		}
	}
	return out, nil
}

// composedKeyOf is a composition's key in the state.
func composedKeyOf(comp delivery.Composed) (string, error) {
	if comp.Module == "" {
		return delivery.ComposedNodeKey(comp.Node)
	}
	return delivery.ComposedKey(comp.Node, comp.Module, comp.FP)
}

// pastKeeping is, of the module keys the state holds, those whose fingerprint is not among the five its module wrote
// last (delivery.ComposedKeep, by revision) and that no want names (ADR 0593 §6). Under pubWrite.
func (c *Composer) pastKeeping() []string {
	c.mu.Lock()
	wanted := map[string]string{}
	for module, w := range c.wants {
		wanted[module] = w.FP
	}
	c.mu.Unlock()
	type kept struct {
		key string
		fp  string
		rev uint64
	}
	byModule := map[string][]kept{}
	for key, e := range c.pubKept {
		_, module, fp, isNode, err := delivery.ParseComposedKey(key)
		if err != nil || isNode {
			continue
		}
		byModule[module] = append(byModule[module], kept{key, fp, e.Revision})
	}
	var out []string
	for module, ks := range byModule {
		sort.Slice(ks, func(i, j int) bool { return ks[i].rev > ks[j].rev })
		for i, k := range ks {
			if i < delivery.ComposedKeep || wanted[module] == k.fp {
				continue
			}
			out = append(out, k.key)
		}
	}
	sort.Strings(out)
	return out
}

// oweComposed keeps the state due and has the loop make a pass again after the growing bound, as owePackages does.
func (c *Composer) oweComposed() string {
	c.mu.Lock()
	defer c.mu.Unlock()
	c.pubDue = true
	c.wait = LaterWait(c.pubTries)
	c.pubTries++
	c.owed.Store(true)
	return fmt.Sprintf("tried again at %s (try %d)", time.Now().Add(c.wait).Format("15:04:05"), c.pubTries)
}

// sayComposed says a line about the state once until it changes; "" clears it.
func (c *Composer) sayComposed(line string) {
	c.mu.Lock()
	same := c.pubSaid == line
	c.pubSaid = line
	c.mu.Unlock()
	if !same && line != "" {
		c.env.Say(line)
	}
}

// ranOf is, of what runs, each module's fingerprint by its composition (the node's own left out).
func ranOf(running []delivery.Composed) map[string]string {
	out := map[string]string{}
	for _, comp := range running {
		if comp.Module != "" && comp.FP != "" && !strings.EqualFold(comp.Module, agent.NodeModule) {
			out[comp.Module] = comp.FP
		}
	}
	return out
}