mesh-host/internal/agent/agent.go

// Package agent is the node agent (novox/hq ADR 0429 §4, ADR 0539): the node-engine's own process that follows the
// node's wants and applies one module at a time.
//
// **It watches, it never polls.** A want `want.<node>.<module>` names a fingerprint and the digest of the one composition
// the assigner allows the node to apply for it; the composition itself waits under `<node>.<module>.<fingerprint>` in
// the node's bucket `composed-<node>`. When both are there and the composition's digest is the want's, and what runs
// differs, the agent applies that module's composition and writes its have with the fingerprint and the digest. A want
// `removed` removes that module's own resources. Nothing is applied for a composition no want names, and a failed apply
// is retried by the next change of either key, never by a clock (research 063 §9.1).
package agent

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"sync"

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

// Update is one change of a key in a watched bucket. Caught marks the end of the keys that were there when the watch
// began; every Update after it is a change.
type Update struct {
	Key      string
	Value    []byte
	Revision uint64
	Deleted  bool
	Caught   bool
}

// Watcher follows one bucket: the keys matching the pattern now, a Caught update, then every change, until ctx ends.
type Watcher interface {
	Watch(ctx context.Context, pattern string) (<-chan Update, error)
}

// NodeModule is the scope of the node's own composition, `<node>.node`.
const NodeModule = "node"

// Have is what a module's have key says, as the agent reads and writes it.
type Have struct {
	Module   string
	FP       string
	Composed string
	// Phase is "running", "failed", "held" or "removed"; empty reads as running.
	Phase  string
	Detail string
}

// Haves is the node's have bucket as the agent uses it.
type Haves interface {
	// Read is what the node-engine last wrote for a module, and whether it wrote anything.
	Read(ctx context.Context, module string) (Have, bool)
	// Write writes a module's have where it differs from the bucket's.
	Write(ctx context.Context, h Have)
	// Removed writes the module's have with the phase removed, once, if a key of it exists.
	Removed(ctx context.Context, module string)
}

// Result is how one apply or removal came out.
type Result struct {
	Failed bool
	Detail string
}

// Backup is the answer of the backup asked before a module moves.
type Backup struct {
	Done  bool
	Words string
}

// Env is everything the agent needs of the node.
type Env struct {
	Node     string
	Wants    Watcher
	Composed Watcher
	Haves    Haves
	// Apply applies one module's composition (scope NodeModule for the node's own), given the compositions the node's
	// other modules run so a path, unit or group they hold is not removed or taken. It is called with one module at a
	// time.
	Apply func(ctx context.Context, c delivery.Composed, others []delivery.Composed) Result
	// Remove removes the module's own resources and nothing of another's.
	Remove func(ctx context.Context, module string, others []delivery.Composed) Result
	// Backup backs a module up before a new composition of it is applied; nil backs up nothing.
	Backup func(ctx context.Context, module string) Backup
	// Kept is the resources the node last applied for a module from the declaration it kept before the switch (ADR 0539
	// §9), as JSON, and whether it held the module. Nil when the node kept none.
	Kept func(module string) ([]json.RawMessage, bool)
	// Verify judges a composition's bytes before anything of it is read: a composition whose signer is not trusted is
	// refused (ADR 0539 §3). Nil trusts the bucket, which only the controller and the operator can write.
	Verify func(key string, raw []byte) error
	Say    func(string)
}

// Agent follows the wants.
type Agent struct {
	env Env

	mu      sync.Mutex
	wants   map[string]wantNow // by module
	comps   map[string]compNow // by composed key
	running map[string]delivery.Have
	// attempt is the last apply of a module that failed or is held, with the revisions it was made from.
	attempt map[string]attempt
	checked map[string]bool
	// prior is a module the node ran before this agent started, so it has something to back up.
	prior map[string]bool
	said  map[string]string
}

type wantNow struct {
	Want    delivery.Want
	Node    bool // the node's own want
	Present bool
	Rev     uint64
}

type compNow struct {
	Raw []byte
	C   delivery.Composed
	Dig string
	Rev uint64
	Err error
}

type attempt struct {
	FP, Composed     string
	WantRev, CompRev uint64
	Held             bool
	AnsweredRev      uint64
}

// New is an agent for a node.
func New(env Env) *Agent {
	if env.Say == nil {
		env.Say = func(string) {}
	}
	return &Agent{env: env, wants: map[string]wantNow{}, comps: map[string]compNow{}, running: map[string]delivery.Have{},
		attempt: map[string]attempt{}, checked: map[string]bool{}, prior: map[string]bool{}, said: map[string]string{}}
}

// Run follows the wants until ctx ends. The keys there when it starts are taken first, the node's own composition
// before any module's, then each change as it comes.
func (a *Agent) Run(ctx context.Context) error {
	wants, err := a.env.Wants.Watch(ctx, "want."+a.env.Node+".>")
	if err != nil {
		return fmt.Errorf("the agent cannot watch its wants: %w", err)
	}
	comps, err := a.env.Composed.Watch(ctx, a.env.Node+".>")
	if err != nil {
		return fmt.Errorf("the agent cannot watch its compositions: %w", err)
	}
	var wantsCaught, compsCaught bool
	dirty := map[string]bool{}
	for {
		select {
		case <-ctx.Done():
			return nil
		case u, ok := <-wants:
			if !ok {
				return fmt.Errorf("the watch of the wants ended")
			}
			if u.Caught {
				wantsCaught = true
			} else if m := a.takeWant(u); m != "" {
				dirty[m] = true
			}
		case u, ok := <-comps:
			if !ok {
				return fmt.Errorf("the watch of the compositions ended")
			}
			if u.Caught {
				compsCaught = true
			} else if m := a.takeComposition(u); m != "" {
				dirty[m] = true
			}
		}
		// Nothing is applied until what was there at the start is all in: a module's composition may arrive after its
		// want, and the node's own composition comes before any module's.
		if !(wantsCaught && compsCaught) || len(dirty) == 0 {
			continue
		}
		a.settle(ctx, dirty)
		dirty = map[string]bool{}
	}
}

// takeWant takes a change of a want key and returns the module it concerns, or "" for a key that is not a want.
func (a *Agent) takeWant(u Update) string {
	a.mu.Lock()
	defer a.mu.Unlock()
	if _, err := delivery.ParseWantNodeKey(u.Key); err == nil {
		w := wantNow{Node: true, Rev: u.Revision}
		if !u.Deleted {
			nw, err := delivery.DecodeNodeWant(u.Value)
			if err != nil {
				a.sayOnce("want:node", fmt.Sprintf("the node's own want is not one: %v", err))
				return ""
			}
			w.Present, w.Want = true, delivery.Want{FP: "", Composed: nw.Composed}
		}
		a.wants[NodeModule] = w
		return NodeModule
	}
	node, module, err := delivery.ParseWantKey(u.Key)
	if err != nil || node != a.env.Node {
		return ""
	}
	w := wantNow{Rev: u.Revision}
	if !u.Deleted {
		want, err := delivery.DecodeWant(u.Value)
		if err != nil {
			a.sayOnce("want:"+module, fmt.Sprintf("the want of %s is not one: %v", module, err))
			return ""
		}
		w.Present, w.Want = true, want
	}
	a.wants[module] = w
	return module
}

// takeComposition takes a change of a composition key and returns the module it concerns.
func (a *Agent) takeComposition(u Update) string {
	node, module, _, isNode, err := delivery.ParseComposedKey(u.Key)
	if err != nil || node != a.env.Node {
		return ""
	}
	a.mu.Lock()
	defer a.mu.Unlock()
	if isNode {
		module = NodeModule
	}
	if u.Deleted {
		delete(a.comps, u.Key)
		return module
	}
	now := compNow{Raw: u.Value, Rev: u.Revision}
	if a.env.Verify != nil {
		if err := a.env.Verify(u.Key, u.Value); err != nil {
			now.Err = fmt.Errorf("refused: %w", err)
			a.comps[u.Key] = now
			a.sayOnce("verify:"+u.Key, fmt.Sprintf("the composition %s is refused and nothing of it is applied: %v", u.Key, err))
			return module
		}
	}
	c, err := delivery.DecodeComposed(u.Value)
	if err == nil {
		now.C = c
		now.Dig, err = delivery.ComposedDigestOf(c)
	}
	now.Err = err
	if err != nil {
		a.sayOnce("decode:"+u.Key, fmt.Sprintf("the composition %s cannot be read, and nothing of it is applied: %v", u.Key, err))
	}
	a.comps[u.Key] = now
	return module
}

func (a *Agent) sayOnce(key, line string) {
	if a.said[key] == line {
		return
	}
	a.said[key] = line
	a.env.Say(line)
}

// settle applies what the dirty modules' wants allow: the node's own first, then each module in name order.
func (a *Agent) settle(ctx context.Context, dirty map[string]bool) {
	order := make([]string, 0, len(dirty))
	for m := range dirty {
		order = append(order, m)
	}
	sort.Slice(order, func(i, j int) bool {
		if (order[i] == NodeModule) != (order[j] == NodeModule) {
			return order[i] == NodeModule
		}
		return order[i] < order[j]
	})
	for _, m := range order {
		if ctx.Err() != nil {
			return
		}
		a.settleOne(ctx, m)
	}
}

func (a *Agent) compositionFor(module, fp string) (compNow, bool) {
	var key string
	var err error
	if module == NodeModule {
		key, err = delivery.ComposedNodeKey(a.env.Node)
	} else {
		key, err = delivery.ComposedKey(a.env.Node, module, fp)
	}
	if err != nil {
		return compNow{}, false
	}
	c, ok := a.comps[key]
	return c, ok
}

// others are the compositions the node's other modules run now, so what they hold is not taken.
func (a *Agent) others(except string) []delivery.Composed {
	var out []delivery.Composed
	mods := make([]string, 0, len(a.running))
	for m := range a.running {
		if m != except {
			mods = append(mods, m)
		}
	}
	sort.Strings(mods)
	for _, m := range mods {
		h := a.running[m]
		if h.Phase == delivery.PhaseRemoved {
			continue
		}
		if c, ok := a.compositionFor(m, h.FP); ok && c.Err == nil && c.Dig == h.Composed {
			out = append(out, c.C)
		}
	}
	return out
}

func (a *Agent) settleOne(ctx context.Context, module string) {
	a.mu.Lock()
	w := a.wants[module]
	a.mu.Unlock()
	if !w.Present {
		return
	}
	if w.Want.Removed {
		a.remove(ctx, module, w)
		return
	}
	a.mu.Lock()
	comp, have := a.compositionFor(module, w.Want.FP)
	a.mu.Unlock()
	if !have || comp.Err != nil {
		return // waits for the composition; a change of its key settles the module again
	}
	if comp.Dig != w.Want.Composed {
		// A composition whose digest no want names is not applied.
		a.mu.Lock()
		a.sayOnce("digest:"+module, fmt.Sprintf("the composition of %s at %s has digest %s, and its want names %s; nothing is applied until they agree",
			module, w.Want.FP, comp.Dig, w.Want.Composed))
		a.mu.Unlock()
		return
	}
	a.mu.Lock()
	if !a.checked[module] {
		a.checked[module] = true
		a.mu.Unlock()
		if a.baseline(ctx, module, w, comp) {
			return
		}
	} else {
		a.mu.Unlock()
	}
	a.mu.Lock()
	if r, ok := a.running[module]; ok && r.FP == w.Want.FP && r.Composed == w.Want.Composed && r.Phase != delivery.PhaseRemoved {
		a.mu.Unlock()
		return
	}
	att, tried := a.attempt[module]
	same := tried && att.FP == w.Want.FP && att.Composed == w.Want.Composed
	if same && !att.Held && att.WantRev == w.Rev && att.CompRev == comp.Rev {
		a.mu.Unlock()
		return // failed already for exactly these keys: retried by the next change of either
	}
	if same && att.Held && att.AnsweredRev >= w.Rev && att.CompRev == comp.Rev {
		a.mu.Unlock()
		return // held for a failed backup, and no answer since
	}
	others := a.others(module)
	a.mu.Unlock()
	a.apply(ctx, module, w, comp, others)
}

// baseline decides whether what runs already is what the want names, once per module at the agent's start: the node's
// own have (what this node-engine last applied), else the declaration it kept before the switch (ADR 0539 §9). True when
// it is, with the have written; nothing is applied.
func (a *Agent) baseline(ctx context.Context, module string, w wantNow, comp compNow) bool {
	if module != NodeModule {
		if h, ok := a.env.Haves.Read(ctx, module); ok {
			a.mu.Lock()
			a.prior[module] = true
			a.mu.Unlock()
			if h.FP == w.Want.FP && h.Composed == w.Want.Composed &&
				(h.Phase == "" || h.Phase == string(delivery.PhaseRunning)) {
				a.mu.Lock()
				a.running[module] = delivery.Have{FP: h.FP, Composed: h.Composed, Phase: delivery.PhaseRunning}
				a.mu.Unlock()
				return true
			}
		}
	}
	if a.env.Kept == nil {
		return false
	}
	kept, ok := a.env.Kept(module)
	if ok {
		a.mu.Lock()
		a.prior[module] = true
		a.mu.Unlock()
	}
	if !ok || !SameResources(kept, comp.C.Resources) {
		return false
	}
	a.mu.Lock()
	a.running[module] = delivery.Have{FP: w.Want.FP, Composed: w.Want.Composed, Phase: delivery.PhaseRunning}
	a.mu.Unlock()
	a.env.Haves.Write(ctx, Have{Module: module, FP: w.Want.FP, Composed: w.Want.Composed, Phase: string(delivery.PhaseRunning)})
	return true
}

func (a *Agent) apply(ctx context.Context, module string, w wantNow, comp compNow, others []delivery.Composed) {
	fp, dig := w.Want.FP, w.Want.Composed
	a.mu.Lock()
	_, ranBefore := a.running[module]
	ranBefore = ranBefore || a.prior[module]
	a.mu.Unlock()

	// A data module is backed up before a new composition of it moves (ADR 0429 §14), unless the want says to go on
	// without one (ADR 0539 §16).
	if comp.C.Data && ranBefore && a.env.Backup != nil && w.Want.Backup != delivery.BackupSkip {
		a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseBackingUp)})
		if got := a.env.Backup(ctx, module); !got.Done {
			a.env.Say(fmt.Sprintf("%s is held: its backup before %s failed: %s", module, fp, got.Words))
			a.mu.Lock()
			a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev, Held: true, AnsweredRev: w.Rev}
			prev := a.running[module]
			a.mu.Unlock()
			// It keeps running what it has: the have says what it holds and why it waits.
			a.env.Haves.Write(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseHeld),
				Detail: got.Words})
			return
		}
	}
	a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseApplying)})
	res := a.env.Apply(ctx, comp.C, others)
	a.mu.Lock()
	if res.Failed {
		a.attempt[module] = attempt{FP: fp, Composed: dig, WantRev: w.Rev, CompRev: comp.Rev}
		a.mu.Unlock()
		a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseFailed), Detail: res.Detail})
		return
	}
	delete(a.attempt, module)
	a.running[module] = delivery.Have{FP: fp, Composed: dig, Phase: delivery.PhaseRunning}
	a.mu.Unlock()
	a.env.Haves.Write(ctx, Have{Module: module, FP: fp, Composed: dig, Phase: string(delivery.PhaseRunning), Detail: res.Detail})
}

func (a *Agent) remove(ctx context.Context, module string, w wantNow) {
	a.mu.Lock()
	att := a.attempt[module]
	r, ran := a.running[module]
	if att.FP == "removed" && att.WantRev == w.Rev {
		a.mu.Unlock()
		return
	}
	if ran && r.Phase == delivery.PhaseRemoved {
		a.mu.Unlock()
		return
	}
	others := a.others(module)
	a.mu.Unlock()
	res := a.env.Remove(ctx, module, others)
	a.mu.Lock()
	if res.Failed {
		a.attempt[module] = attempt{FP: "removed", WantRev: w.Rev}
		a.mu.Unlock()
		a.env.Say(fmt.Sprintf("removing %s failed: %s", module, res.Detail))
		return
	}
	delete(a.attempt, module)
	a.running[module] = delivery.Have{Phase: delivery.PhaseRemoved}
	a.mu.Unlock()
	a.env.Haves.Removed(ctx, module)
}

// SameResources says whether two lists of resources hold the same resources, each by its id and its canonical JSON.
func SameResources(a, b []json.RawMessage) bool {
	ma, ok := canonByID(a)
	if !ok {
		return false
	}
	mb, ok := canonByID(b)
	if !ok || len(ma) != len(mb) {
		return false
	}
	for id, x := range ma {
		if y, there := mb[id]; !there || x != y {
			return false
		}
	}
	return true
}

func canonByID(rs []json.RawMessage) (map[string]string, bool) {
	out := map[string]string{}
	for _, raw := range rs {
		var head struct {
			ID string `json:"id"`
		}
		if json.Unmarshal(raw, &head) != nil || head.ID == "" {
			return nil, false
		}
		canon, err := Canonical(raw)
		if err != nil {
			return nil, false
		}
		out[head.ID] = string(canon)
	}
	return out, true
}

// Canonical is raw with its object keys sorted and its spacing dropped, its numbers kept as written.
func Canonical(raw json.RawMessage) (json.RawMessage, error) {
	var v any
	d := json.NewDecoder(bytes.NewReader(raw))
	d.UseNumber()
	if err := d.Decode(&v); err != nil {
		return nil, err
	}
	return json.Marshal(v)
}