From d5dd7761c6a86ca4557a1025644e9d661a99d510 Mon Sep 17 00:00:00 2001 From: jochen Date: Sat, 24 Oct 2026 16:36:21 +0200 Subject: [PATCH] The node agent: follows the wants and the node's compositions, applies one module at a time, backs a data module up first, holds it on a failed backup (R6-A A1, A2, A5, A6) --- cmd/mesh-host/main.go | 4 - internal/agent/agent.go | 528 +++++++++++++++++++++++++++++++++++ internal/agent/agent_test.go | 485 ++++++++++++++++++++++++++++++++ internal/link/have.go | 93 +++++- internal/link/messages.go | 17 +- 5 files changed, 1109 insertions(+), 18 deletions(-) create mode 100644 internal/agent/agent.go create mode 100644 internal/agent/agent_test.go diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 6572167..94ed226 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1915,10 +1915,6 @@ func sayFacts(ctx context.Context, opts options, signer ed25519.PublicKey, sessi held[h.ID] = true } report := factsOf(ctx, opts, declared, known, apply.Report{}, held) - report.Modules = moduleFacts(declared, apply.Report{}, nil) - if report.Modules == nil { - report.Modules = []link.ModuleFact{} - } session.SayFacts(ctx, report) } diff --git a/internal/agent/agent.go b/internal/agent/agent.go new file mode 100644 index 0000000..471b0d7 --- /dev/null +++ b/internal/agent/agent.go @@ -0,0 +1,528 @@ +// 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" + "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) + // 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) +} diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go new file mode 100644 index 0000000..555c583 --- /dev/null +++ b/internal/agent/agent_test.go @@ -0,0 +1,485 @@ +package agent + +import ( + "context" + "encoding/json" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +const node = "n1" + +func fpOf(c string) string { return delivery.SourcePrefix2 + strings.Repeat(c, 64) } + +// A watcher the test writes to; it says Caught once the keys written before the watch began are out. +type feed struct { + mu sync.Mutex + ch chan Update + rev uint64 + first []Update +} + +func newFeed() *feed { return &feed{ch: make(chan Update, 256)} } + +func (f *feed) Watch(_ context.Context, _ string) (<-chan Update, error) { + f.mu.Lock() + defer f.mu.Unlock() + for _, u := range f.first { + f.ch <- u + } + f.ch <- Update{Caught: true} + return f.ch, nil +} + +// before writes a key as if it was there when the watch began. +func (f *feed) before(key string, value []byte) { + f.mu.Lock() + defer f.mu.Unlock() + f.rev++ + f.first = append(f.first, Update{Key: key, Value: value, Revision: f.rev}) +} + +func (f *feed) put(key string, value []byte) { + f.mu.Lock() + f.rev++ + r := f.rev + f.mu.Unlock() + f.ch <- Update{Key: key, Value: value, Revision: r} +} + +type haves struct { + mu sync.Mutex + stored map[string]Have + log []Have + removed []string +} + +func (h *haves) Read(_ context.Context, module string) (Have, bool) { + h.mu.Lock() + defer h.mu.Unlock() + x, ok := h.stored[module] + return x, ok +} +func (h *haves) Write(_ context.Context, x Have) { + h.mu.Lock() + defer h.mu.Unlock() + if h.stored == nil { + h.stored = map[string]Have{} + } + h.stored[x.Module] = x + h.log = append(h.log, x) +} +func (h *haves) Removed(_ context.Context, m string) { + h.mu.Lock() + defer h.mu.Unlock() + h.removed = append(h.removed, m) +} +func (h *haves) last(module string) Have { + h.mu.Lock() + defer h.mu.Unlock() + return h.stored[module] +} + +type bed struct { + t *testing.T + wants *feed + comps *feed + haves *haves + mu sync.Mutex + applied []string + removed []string + backups []string + failWith string + backup func(module string) Backup + kept map[string][]json.RawMessage + env Env +} + +func newBed(t *testing.T) *bed { + b := &bed{t: t, wants: newFeed(), comps: newFeed(), haves: &haves{}} + b.env = Env{Node: node, Wants: b.wants, Composed: b.comps, Haves: b.haves, + Apply: func(_ context.Context, c delivery.Composed, _ []delivery.Composed) Result { + b.mu.Lock() + defer b.mu.Unlock() + name := c.Module + if name == "" { + name = NodeModule + } + b.applied = append(b.applied, name) + if b.failWith != "" { + return Result{Failed: true, Detail: b.failWith} + } + return Result{} + }, + Remove: func(_ context.Context, m string, _ []delivery.Composed) Result { + b.mu.Lock() + defer b.mu.Unlock() + b.removed = append(b.removed, m) + return Result{} + }, + Backup: func(_ context.Context, m string) Backup { + b.mu.Lock() + defer b.mu.Unlock() + b.backups = append(b.backups, m) + if b.backup != nil { + return b.backup(m) + } + return Backup{Done: true} + }, + Kept: func(m string) ([]json.RawMessage, bool) { + r, ok := b.kept[m] + return r, ok + }, + Say: func(s string) { t.Log(s) }, + } + return b +} + +func (b *bed) run() (stop func()) { + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan struct{}) + go func() { defer close(done); _ = New(b.env).Run(ctx) }() + return func() { cancel(); <-done } +} + +func file(module, id, content string) json.RawMessage { + raw, _ := json.Marshal(map[string]any{"id": module + "." + id, "type": "file", "path": "/tmp/r6/" + module + "/" + id, + "content": content, "module": module}) + return raw +} + +// composition is a module's composition at a fingerprint: its key, its bytes and its digest, from the writer's code. +func composition(t *testing.T, module, fp string, data bool, rs ...json.RawMessage) (key string, raw []byte, digest string) { + t.Helper() + c := delivery.Composed{Node: node, Module: module, FP: fp, Data: data, Resources: rs} + var err error + if module == NodeModule { + c.Module, c.FP = "", "" + key, err = delivery.ComposedNodeKey(node) + } else { + key, err = delivery.ComposedKey(node, module, fp) + } + if err != nil { + t.Fatal(err) + } + if raw, err = delivery.EncodeComposed(c); err != nil { + t.Fatal(err) + } + if digest, err = delivery.ComposedDigestOf(c); err != nil { + t.Fatal(err) + } + return key, raw, digest +} + +func wantFor(t *testing.T, module, fp, digest, backup string) (string, []byte) { + t.Helper() + key, err := delivery.WantKey(node, module) + if err != nil { + t.Fatal(err) + } + raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Composed: digest, Backup: backup}) + if err != nil { + t.Fatal(err) + } + return key, raw +} + +func eventually(t *testing.T, what string, ok func() bool) { + t.Helper() + deadline := time.Now().Add(3 * time.Second) + for !ok() { + if time.Now().After(deadline) { + t.Fatal(what) + } + time.Sleep(5 * time.Millisecond) + } +} + +func (b *bed) appliedList() []string { + b.mu.Lock() + defer b.mu.Unlock() + return append([]string(nil), b.applied...) +} + +// R6-A AC1 (the agent, with a bus that is a fake): given a want naming a digest and that composition, the module is +// applied and its have names the fingerprint and the digest; a composition whose digest no want names is not applied. +func TestAWantAndItsCompositionApplyTheModuleAndTheHaveNamesBoth(t *testing.T) { + b := newBed(t) + stop := b.run() + defer stop() + + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + // A composition no want names yet is not applied. + b.comps.put(ck, craw) + time.Sleep(100 * time.Millisecond) + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("a composition no want names was applied: %v", got) + } + wk, wraw := wantFor(t, "web", fp, dig, "") + b.wants.put(wk, wraw) + eventually(t, "the module was not applied", func() bool { return len(b.appliedList()) == 1 }) + eventually(t, "the have does not name both", func() bool { + h := b.haves.last("web") + return h.FP == fp && h.Composed == dig && h.Phase == string(delivery.PhaseRunning) + }) + + // A want naming a digest the composition at that fingerprint does not have: nothing is applied. + fp2 := fpOf("b") + ck2, craw2, _ := composition(t, "web", fp2, false, file("web", "x", "2")) + b.comps.put(ck2, craw2) + wk2, wraw2 := wantFor(t, "web", fp2, "sha256:"+strings.Repeat("0", 64), "") + b.wants.put(wk2, wraw2) + time.Sleep(100 * time.Millisecond) + if got := b.appliedList(); len(got) != 1 { + t.Fatalf("a want naming another digest applied the composition: %v", got) + } +} + +// R6-A AC2: given a kept declaration and wants and compositions equal to it, when the agent starts, nothing is applied +// and every have names its fingerprint and digest. +func TestStartingOnWhatTheKeptDeclarationHoldsAppliesNothing(t *testing.T) { + b := newBed(t) + fp := fpOf("a") + rs := []json.RawMessage{file("web", "x", "1"), file("web", "y", "2")} + ck, craw, dig := composition(t, "web", fp, false, rs...) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.comps.before(ck, craw) + b.wants.before(wk, wraw) + // The kept declaration holds the same resources, in another order and spacing. + b.kept = map[string][]json.RawMessage{"web": {json.RawMessage(strings.ReplaceAll(string(rs[1]), ",", " , ")), rs[0]}} + stop := b.run() + defer stop() + eventually(t, "the have was not written", func() bool { return b.haves.last("web").Composed == dig }) + if h := b.haves.last("web"); h.FP != fp || h.Phase != string(delivery.PhaseRunning) { + t.Fatalf("the have: %+v", h) + } + time.Sleep(50 * time.Millisecond) + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("a module that already runs what the want names was applied: %v", got) + } +} + +// What the node-engine itself last applied is what runs after a restart, not the kept declaration. +func TestStartingOnItsOwnHaveAppliesNothing(t *testing.T) { + b := newBed(t) + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.haves.stored = map[string]Have{"web": {Module: "web", FP: fp, Composed: dig, Phase: "running"}} + b.comps.before(ck, craw) + b.wants.before(wk, wraw) + stop := b.run() + defer stop() + time.Sleep(150 * time.Millisecond) + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("a module that runs what the want names was applied after a restart: %v", got) + } +} + +// A module whose kept resources differ from the composition is applied, and a failed apply is retried only by a change +// of either key. +func TestAFailedApplyIsRetriedOnlyByAChangeOfEitherKey(t *testing.T) { + b := newBed(t) + b.failWith = "web.x: no space" + stop := b.run() + defer stop() + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.comps.put(ck, craw) + b.wants.put(wk, wraw) + eventually(t, "not applied", func() bool { return len(b.appliedList()) == 1 }) + eventually(t, "the have does not say failed", func() bool { + h := b.haves.last("web") + return h.Phase == string(delivery.PhaseFailed) && strings.Contains(h.Detail, "no space") && h.FP == fp + }) + // An unrelated key changes: not a retry. + ck9, craw9, _ := composition(t, "web", fpOf("9"), false, file("web", "x", "9")) + b.comps.put(ck9, craw9) + time.Sleep(150 * time.Millisecond) + if n := len(b.appliedList()); n != 1 { + t.Fatalf("a failed apply was retried by an unrelated key (%d applies)", n) + } + // The want is written again: the retry. + b.mu.Lock() + b.failWith = "" + b.mu.Unlock() + b.wants.put(wk, wraw) + eventually(t, "the change of the want did not retry", func() bool { return len(b.appliedList()) == 2 }) + eventually(t, "not running", func() bool { return b.haves.last("web").Phase == string(delivery.PhaseRunning) }) +} + +// R6-A AC5: a data module that ran before asks its backup before any resource changes; a refusal holds it, nothing is +// applied and the have says held; when the want then says skip, the module is applied. +func TestADataModuleIsBackedUpFirstAndHeldWhenItFails(t *testing.T) { + b := newBed(t) + fp1, fp2 := fpOf("a"), fpOf("b") + ck1, craw1, dig1 := composition(t, "db", fp1, true, file("db", "x", "1")) + wk1, wraw1 := wantFor(t, "db", fp1, dig1, "") + b.comps.before(ck1, craw1) + b.wants.before(wk1, wraw1) + b.kept = map[string][]json.RawMessage{"db": {file("db", "x", "1")}} + refuse := true + b.backup = func(string) Backup { + if refuse { + return Backup{Words: "restic: the repository is locked"} + } + return Backup{Done: true} + } + stop := b.run() + defer stop() + eventually(t, "baseline", func() bool { return b.haves.last("db").Composed == dig1 }) + + ck2, craw2, dig2 := composition(t, "db", fp2, true, file("db", "x", "2")) + b.comps.put(ck2, craw2) + wk2, wraw2 := wantFor(t, "db", fp2, dig2, "") + b.wants.put(wk2, wraw2) + eventually(t, "the backup was not asked", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.backups) == 1 + }) + eventually(t, "the have does not say held", func() bool { + h := b.haves.last("db") + return h.Phase == string(delivery.PhaseHeld) && strings.Contains(h.Detail, "repository is locked") && h.FP == fp1 + }) + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("something was applied behind a failed backup: %v", got) + } + + // The answer arrives in the module's own want: back up again (it fails again), then skip. + wk, again := wantFor(t, "db", fp2, dig2, delivery.BackupAgain) + b.wants.put(wk, again) + eventually(t, "back up again did not ask the backup", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.backups) == 2 + }) + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("applied behind a second failed backup: %v", got) + } + _, skip := wantFor(t, "db", fp2, dig2, delivery.BackupSkip) + b.wants.put(wk, skip) + eventually(t, "skip did not apply the module", func() bool { return len(b.appliedList()) == 1 }) + b.mu.Lock() + n := len(b.backups) + b.mu.Unlock() + if n != 2 { + t.Fatalf("a backup was asked after the answer was to skip: %d asks", n) + } + eventually(t, "not running at the new fingerprint", func() bool { + h := b.haves.last("db") + return h.Phase == string(delivery.PhaseRunning) && h.FP == fp2 && h.Composed == dig2 + }) +} + +// A data module with a good backup is applied after it; one that never ran has nothing to back up. +func TestABackupDoneIsFollowedByTheApplyAndANewModuleIsNotBackedUp(t *testing.T) { + b := newBed(t) + stop := b.run() + defer stop() + fp := fpOf("c") + ck, craw, dig := composition(t, "db", fp, true, file("db", "x", "1")) + wk, wraw := wantFor(t, "db", fp, dig, "") + b.comps.put(ck, craw) + b.wants.put(wk, wraw) + eventually(t, "not applied", func() bool { return len(b.appliedList()) == 1 }) + b.mu.Lock() + n := len(b.backups) + b.mu.Unlock() + if n != 0 { + t.Fatalf("a module that never ran was backed up (%d)", n) + } + fp2 := fpOf("d") + ck2, craw2, dig2 := composition(t, "db", fp2, true, file("db", "x", "2")) + wk2, wraw2 := wantFor(t, "db", fp2, dig2, "") + b.comps.put(ck2, craw2) + b.wants.put(wk2, wraw2) + eventually(t, "second not applied", func() bool { return len(b.appliedList()) == 2 }) + b.mu.Lock() + n = len(b.backups) + b.mu.Unlock() + if n != 1 { + t.Fatalf("a module that ran before was backed up %d times, want once", n) + } +} + +// R6-A AC4 at the agent: a removed want removes that module only, and its have is written removed. +func TestARemovedWantRemovesThatModuleOnly(t *testing.T) { + b := newBed(t) + stop := b.run() + defer stop() + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + b.comps.put(ck, craw) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.wants.put(wk, wraw) + eventually(t, "not applied", func() bool { return len(b.appliedList()) == 1 }) + gone, _ := delivery.EncodeWant(delivery.Want{Removed: true}) + b.wants.put(wk, gone) + eventually(t, "not removed", func() bool { + b.mu.Lock() + defer b.mu.Unlock() + return len(b.removed) == 1 && b.removed[0] == "web" + }) + eventually(t, "the have was not written removed", func() bool { + b.haves.mu.Lock() + defer b.haves.mu.Unlock() + return len(b.haves.removed) == 1 + }) +} + +// The node's own composition is applied before any module's, whichever key came first. +func TestTheNodesOwnCompositionComesBeforeAnyModules(t *testing.T) { + b := newBed(t) + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + wk, wraw := wantFor(t, "web", fp, dig, "") + nk, nraw, ndig := composition(t, NodeModule, "", false) + nwk, err := delivery.WantNodeKey(node) + if err != nil { + t.Fatal(err) + } + nwraw, err := delivery.EncodeNodeWant(delivery.NodeWant{Composed: ndig}) + if err != nil { + t.Fatal(err) + } + b.comps.before(ck, craw) + b.comps.before(nk, nraw) + b.wants.before(wk, wraw) + b.wants.before(nwk, nwraw) + stop := b.run() + defer stop() + eventually(t, "both not applied", func() bool { return len(b.appliedList()) == 2 }) + if got := b.appliedList(); got[0] != NodeModule || got[1] != "web" { + t.Fatalf("the order was %v, the node's own first", got) + } +} + +// A composition the verifier refuses is never applied (the hook of ADR 0539 §3). +func TestARefusedCompositionIsNeverApplied(t *testing.T) { + b := newBed(t) + b.env.Verify = func(string, []byte) error { return errRefused } + stop := b.run() + defer stop() + fp := fpOf("a") + ck, craw, dig := composition(t, "web", fp, false, file("web", "x", "1")) + b.comps.put(ck, craw) + wk, wraw := wantFor(t, "web", fp, dig, "") + b.wants.put(wk, wraw) + time.Sleep(150 * time.Millisecond) + if got := b.appliedList(); len(got) != 0 { + t.Fatalf("a refused composition was applied: %v", got) + } +} + +type refusal string + +func (r refusal) Error() string { return string(r) } + +var errRefused error = refusal("signed by a key this node does not trust") diff --git a/internal/link/have.go b/internal/link/have.go index 702ca78..db350cd 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -37,6 +37,8 @@ type haveState struct { modules map[string]ModuleFact health map[string]delivery.Health node *delivery.HaveNode + // nodeComposed is the digest of the node's own composition it runs (ADR 0539 §9), kept beside the node's facts. + nodeComposed string // written is the build of each module this node-engine wrote a have key for and that the declaration it applied // still holds; removed is each such module the latest declaration no longer holds, with the build its key last // said, until the key is written once with the phase removed (issue 1369, novox/hq ADR 0408 §4: never deleted). @@ -98,10 +100,22 @@ func nodeFacts(r Report) delivery.HaveNode { // noteApply takes what an apply's report says of the modules and the node. func (q *Session) noteApply(r Report) bool { - if r.Modules == nil { - return false - } s := &q.haves + if r.Modules == nil { + // Facts that say nothing of any module (the node agent writes those, ADR 0539 §9): the node's own key and the + // health the facts carry. + if r.Refused != "" { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + if r.Health != nil { + s.health = moduleHealths(r.Health) + } + n := nodeFacts(r) + s.node = &n + return true + } s.mu.Lock() defer s.mu.Unlock() next := map[string]ModuleFact{} @@ -208,9 +222,12 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { sort.Strings(modules) for _, module := range modules { f := s.modules[module] - h := delivery.Have{V: delivery.Version, Build: f.Build, Phase: delivery.PhaseRunning, - Health: delivery.HealthUnknown, Detail: f.Detail} - if f.Failed { + h := delivery.Have{V: delivery.Version, FP: f.FP, Composed: f.Composed, Build: f.Build, + Phase: delivery.PhaseRunning, Health: delivery.HealthUnknown, Detail: f.Detail} + switch { + case f.Phase != "": + h.Phase = delivery.Phase(f.Phase) + case f.Failed: h.Phase = delivery.PhaseFailed } if got, known := s.health[module]; known { @@ -244,8 +261,10 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { key, err := delivery.HaveNodeKey(node) if err == nil { var raw []byte - if raw, err = delivery.EncodeHaveNode(*s.node); err == nil { - want := *s.node + facts := *s.node + facts.Composed = s.nodeComposed + if raw, err = delivery.EncodeHaveNode(facts); err == nil { + want := facts q.putHave(write, bucket, key, raw, func(current []byte) bool { c, err := delivery.DecodeHaveNode(current) return err == nil && delivery.SameHaveNode(c, want) @@ -343,3 +362,61 @@ func (q *Session) haveWritten() { q.HaveWritten() } } + +// NoteModule takes how one module's apply came out and writes its have where it differs from the bucket's (novox/hq ADR +// 0539 §9): the node agent's word for a module, with the fingerprint and the digest it runs. It keeps the module's last +// health, which a look changes without an apply. +func (q *Session) NoteModule(ctx context.Context, f ModuleFact) { + q.init() + s := &q.haves + s.mu.Lock() + if s.modules == nil { + s.modules = map[string]ModuleFact{} + } + s.modules[f.Module] = f + delete(s.removed, f.Module) + s.mu.Unlock() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + q.writeHaves(ctx, bus) +} + +// NoteRemoved takes that a module's own resources were removed, and writes its have with the phase removed once (issue +// 1369): the key is never deleted. +func (q *Session) NoteRemoved(ctx context.Context, module, build string) { + q.init() + s := &q.haves + s.mu.Lock() + if s.removed == nil { + s.removed = map[string]string{} + } + if f, ok := s.modules[module]; ok && build == "" { + build = f.Build + } + s.removed[module] = build + delete(s.modules, module) + delete(s.written, module) + s.mu.Unlock() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + q.writeHaves(ctx, bus) +} + +// NoteNode takes the digest of the node's own composition it runs, and writes the node's key where it differs. +func (q *Session) NoteNode(ctx context.Context, composed string) { + q.init() + s := &q.haves + s.mu.Lock() + s.nodeComposed = composed + if s.node == nil { + n := delivery.HaveNode{V: delivery.Version} + s.node = &n + } + s.mu.Unlock() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + q.writeHaves(ctx, bus) +} diff --git a/internal/link/messages.go b/internal/link/messages.go index 548dafd..3aaba65 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -731,11 +731,16 @@ type FoundFirewall struct { RetiredBy string `json:"retired_by,omitempty"` } -// ModuleFact is how one module's part of an apply came out: the build the node was sent for it, and whether every -// resource of it applied. Detail names the failing resource and its error (issue 1132). +// ModuleFact is how one module's part of an apply came out (novox/hq ADR 0539 §9): the fingerprint FP the module runs and +// the digest Composed of the composition applied for it, beside the build the node was sent, the Phase it is in (empty +// is running, or failed when Failed), and whether every resource of it applied. Detail names the failing resource and its +// error (issue 1132). type ModuleFact struct { - Module string - Build string - Failed bool - Detail string + Module string + Build string + FP string + Composed string + Phase string + Failed bool + Detail string }