// 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..` names a fingerprint and the digest of the one composition // the assigner allows the node to apply for it; the composition itself waits under `..` in // the node's bucket `composed-`. 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" "crypto/ed25519" "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`. 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) // Trusted is the public keys a composition may be signed by: the controller's and the operator's (ADR 0539 §3). A // composition unsigned, signed by another key, or changed after it was signed is refused and nothing of it is // applied. Nil trusts no key, so nothing is applied. Trusted func() []ed25519.PublicKey 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 retry chan []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{}, retry: make(chan []string, 8)} } // 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 modules := <-a.retry: // Applied again though the keys did not change: a container that waited because it said busy (ADR 0360 // §11) is replaced once it is idle. a.mu.Lock() for _, m := range modules { delete(a.running, m) delete(a.attempt, m) dirty[m] = true } a.mu.Unlock() 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{} } } // Retry asks for the modules to be applied again from the keys they have, though nothing changed: what ADR 0360 §11 // needs for a container that waits because it said busy. It never blocks; a retry asked while eight wait is dropped, and // asked again by its caller's next look. func (a *Agent) Retry(modules ...string) { select { case a.retry <- modules: default: } } // Running is the compositions the node runs now: each module's that the agent applied or found already running, and // the node's own. What the node judges its health by and says of itself is built from them. func (a *Agent) Running() []delivery.Composed { a.mu.Lock() defer a.mu.Unlock() mods := make([]string, 0, len(a.running)) for m := range a.running { mods = append(mods, m) } sort.Strings(mods) var out []delivery.Composed 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 } // 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} c, err := delivery.DecodeComposed(u.Value) if err == nil { var trusted []ed25519.PublicKey if a.env.Trusted != nil { trusted = a.env.Trusted() } // The signature before anything of the composition is used: its digest is what the want names. now.C = c now.Dig, err = delivery.VerifyComposed(c, trusted) if err != nil { now.C = delivery.Composed{} a.sayOnce("verify:"+u.Key, fmt.Sprintf("the composition %s is refused and nothing of it is applied: %v", u.Key, err)) err = fmt.Errorf("refused: %w", err) } } else { a.sayOnce("decode:"+u.Key, fmt.Sprintf("the composition %s cannot be read, and nothing of it is applied: %v", u.Key, err)) } now.Err = 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.Say(fmt.Sprintf("applying %s at %s failed: %s", module, fp, res.Detail)) 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) }