From 02d9d7b0538f1dc8bbda5f38e80c4ea9f8738bd3 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 26 Oct 2026 18:44:58 +0100 Subject: [PATCH] =?UTF-8?q?The=20node=20composes=20its=20own=20modules=20o?= =?UTF-8?q?nce=20the=20catalogue=20places=20it:=20one=20composer=20per=20n?= =?UTF-8?q?ode=20drains=20a=20mark=20set=20by=20any=20input's=20change,=20?= =?UTF-8?q?the=20agent=20applies=20what=20it=20composed=20when=20it=20diff?= =?UTF-8?q?ers,=20and=20a=20record=20move=20waits=20for=20a=20yes=20(issue?= =?UTF-8?q?=201126,=20ADR=200547=20=C2=A72,=20=C2=A73)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- cmd/mesh-host/nodeagent.go | 43 +- go.mod | 2 +- go.sum | 2 + internal/agent/agent.go | 89 +++- internal/agent/agent_test.go | 76 ++++ internal/link/have.go | 3 + internal/link/messages.go | 3 + internal/selfcompose/selfcompose.go | 494 +++++++++++++++++++++++ internal/selfcompose/selfcompose_test.go | 159 ++++++++ 9 files changed, 866 insertions(+), 5 deletions(-) create mode 100644 internal/selfcompose/selfcompose.go create mode 100644 internal/selfcompose/selfcompose_test.go diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index d4e69ab..d6b3fe3 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -13,11 +13,14 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/declaration" "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/overlay" "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/apply" "github.com/novox/mesh-host/internal/link" + "github.com/novox/mesh-host/internal/selfcompose" "github.com/novox/mesh-host/internal/store" ) @@ -73,7 +76,7 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) { phase = "" } h.session.NoteModule(ctx, link.ModuleFact{Module: x.Module, FP: x.FP, Composed: x.Composed, Phase: phase, - Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail)}) + Failed: phase == string(delivery.PhaseFailed), Detail: shorten(x.Detail), Judgement: x.Judgement, Waits: x.Waits}) } func (h sessionHaves) Removed(ctx context.Context, module string) { @@ -476,6 +479,11 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess return res }, } + // The node composes its own modules once the catalogue places it (novox/hq ADR 0547 §2, ADR 0549, issue 1126): its + // placement key is read at the agent's start; until then it applies the controller's compositions as before. + if composer := selfComposer(linked, session, node, say); composer != nil { + env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy + } ag = agent.New(env) current.set(ag) err = ag.Run(linked) @@ -756,3 +764,36 @@ const applyBound = 30 * time.Minute func applyContext(ctx context.Context) (context.Context, context.CancelFunc) { return context.WithTimeout(context.WithoutCancel(ctx), applyBound) } + +// selfComposer is the node's own composer when the catalogue places the node (its key `placement.` in the +// indexer's state), and nil, said once, while it does not or the key cannot be read. +func selfComposer(ctx context.Context, session *link.Session, node string, say func(string)) *selfcompose.Composer { + key, err := delivery.PlacementKey(node) + if err != nil { + say("this node applies the controller's compositions: " + err.Error()) + return nil + } + got, err := session.KVHistory(ctx, delivery.PlacementBucket, key) + if err != nil || len(got) == 0 || got[len(got)-1].Deleted { + why := "the catalogue does not place it yet" + if err != nil { + why = "its placement cannot be read: " + err.Error() + } + say("this node applies the controller's compositions: " + why) + return nil + } + say("this node composes its own modules: the catalogue places it (" + key + ")") + return selfcompose.New(selfcompose.Env{Node: node, + Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { + return bucketWatch{session, bucket}.Watch(ctx, pattern) + }, + Get: func(ctx context.Context, bucket, key string) ([]byte, error) { + h, err := session.KVHistory(ctx, bucket, key) + if err != nil || len(h) == 0 || h[len(h)-1].Deleted { + return nil, err + } + return h[len(h)-1].Value, nil + }, + Mesh: frombus.Mesh{InternalSuffix: overlay.Suffix()}, + Say: say}) +} diff --git a/go.mod b/go.mod index 7d5cc53..87c7994 100644 --- a/go.mod +++ b/go.mod @@ -3,7 +3,7 @@ module github.com/novox/mesh-host go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026173336-9789c8379a74 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026174406-6df0dfe208c6 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 golang.org/x/crypto v0.57.0 diff --git a/go.sum b/go.sum index cb2ad74..4861dd1 100644 --- a/go.sum +++ b/go.sum @@ -4,6 +4,8 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026172945-2b0689f6546f h1:c0N990Te git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026172945-2b0689f6546f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026173336-9789c8379a74 h1:+LrIwBeG0eCN6xOS9jhzL6ag0ipXHZ1vfCFEJqtT1eI= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026173336-9789c8379a74/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026174406-6df0dfe208c6 h1:sIagBHevPfpISBRK2r5EemnhHEGubdfEx/l3K6ujomo= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026174406-6df0dfe208c6/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 467e21b..ea2f8df 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -23,6 +23,7 @@ import ( "sync" "time" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -49,9 +50,13 @@ type Have struct { Module string FP string Composed string - // Phase is "running", "failed", "held" or "removed"; empty reads as running. + // Phase is "running", "failed", "held", "waiting-yes" or "removed"; empty reads as running. Phase string Detail string + // Judgement and Waits are, in phase waiting-yes, the judgement of the move and the digest of the composition that + // waits for a person's yes (ADR 0547 §3). + Judgement string + Waits string } // Haves is the node's have bucket as the agent uses it. @@ -120,6 +125,14 @@ type Env struct { // transient failure only on RetryTransient or a change of either key. Backoff func(try int) time.Duration Say func(string) + // SelfComposed says the compositions are the node's own (novox/hq ADR 0547 §2, issue 1126): Composed hands what the + // node composed itself, unsigned, and a want names the build alone. The composition composed for the want's build is + // the one applied, when it differs from what runs. + SelfComposed bool + // Policy is a module's upgrade policy as its manifest derives it (catalogue.DerivedUpgrade); with SelfComposed, a + // move of a module of policy record that recreates a container or moves its data waits for a person's yes in its + // want (ADR 0547 §3). Nil judges no move. + Policy func(module string) string } // transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has @@ -366,7 +379,14 @@ func (a *Agent) takeComposition(u Update) string { } now := compNow{Raw: u.Value, Rev: u.Revision} c, err := delivery.DecodeComposed(u.Value) - if err == nil { + if err == nil && a.env.SelfComposed { + // The node composed it itself: nothing crossed the bus, so nothing is signed (ADR 0547 §2). + now.C = c + now.Dig, err = delivery.ComposedDigestOf(c) + if err != nil { + now.C = delivery.Composed{} + } + } else if err == nil { var trusted []ed25519.PublicKey if a.env.Trusted != nil { trusted = a.env.Trusted() @@ -636,7 +656,16 @@ func (a *Agent) settleOne(ctx context.Context, module string) { a.mu.Lock() comp, have := a.compositionFor(module, w.Want.FP) a.mu.Unlock() - if (!have || comp.Err != nil || comp.Dig != w.Want.Composed) && w.Want.Composed != "" { + if a.env.SelfComposed { + if !have || comp.Err != nil { + return // waits for the node's own composition of the want's build + } + // The composition the node composed for the want's build is the one it applies (ADR 0547 §2). + w.Want.Composed = comp.Dig + if a.waitsForYes(ctx, module, w, comp) { + return + } + } else if (!have || comp.Err != nil || comp.Dig != w.Want.Composed) && w.Want.Composed != "" { // The composition the want names may be an older revision of the key, replaced by a newer one at the same // fingerprint (a want held there keeps its digest): taken from the revisions the bucket keeps (ADR 0539 clause 1). if old, found := a.revisionNamed(ctx, module, w.Want.FP, w.Want.Composed); found { @@ -925,3 +954,57 @@ func short(s string) string { } return s } + +// waitsForYes judges a move the node composed for a module of policy record (ADR 0547 §3, ADR 0377 §3): one that +// recreates a container or moves the module's data, or that cannot be judged, is not applied while the want names no yes +// for this composition; the have says phase waiting-yes with the judgement, and the assigner asks a person. A composition +// equal to what runs is no move. True when the module waits. +func (a *Agent) waitsForYes(ctx context.Context, module string, w wantNow, comp compNow) bool { + if module == NodeModule || a.env.Policy == nil || a.env.Policy(module) != catalogue.PolicyRecord || w.Want.Yes == comp.Dig { + return false + } + a.mu.Lock() + r, ran := a.running[module] + if ran && r.Composed == comp.Dig && r.Phase != delivery.PhaseRemoved { + a.mu.Unlock() + return false + } + from, known := a.runningComposition(module) + prior := ran || a.prior[module] + a.mu.Unlock() + var j delivery.Judged + switch { + case known: + j = delivery.JudgeMove(&from, comp.C) + case prior: + j = delivery.Judged{Judgement: delivery.JudgedCannot, Words: "what " + module + " runs on this node cannot be read"} + default: + j = delivery.JudgeMove(nil, comp.C) + } + switch j.Judgement { + case delivery.JudgedRecreates, delivery.JudgedMovesData, delivery.JudgedCannot: + default: + return false + } + a.mu.Lock() + a.sayOnce("yes:"+module+":"+comp.Dig, fmt.Sprintf("%s at %s waits for a person's yes: %s", module, w.Want.FP, j.Words)) + prev := a.running[module] + a.mu.Unlock() + a.env.Haves.Write(ctx, Have{Module: module, FP: prev.FP, Composed: prev.Composed, Phase: string(delivery.PhaseWaitingYes), + Detail: j.Words, Judgement: string(j.Judgement), Waits: comp.Dig}) + return true +} + +// runningComposition is what the module runs now: the composition last applied, else its resources in the declaration +// the node kept from before the switch. Called with a.mu held. +func (a *Agent) runningComposition(module string) (delivery.Composed, bool) { + if c, ok := a.applied[module]; ok && c.Err == nil { + return c.C, true + } + if a.env.Kept != nil { + if rs, ok := a.env.Kept()[module]; ok { + return delivery.Composed{Node: a.env.Node, Module: module, Resources: rs}, true + } + } + return delivery.Composed{}, false +} diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 55d61b1..91c167b 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -10,6 +10,7 @@ import ( "testing" "time" + "git.novox.be/novox/mesh-sdk/go/catalogue" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -952,3 +953,78 @@ func TestAModuleRunningFromTheKeptDeclarationKeepsItsFilledPaths(t *testing.T) { t.Fatalf("what runs from the kept declaration: %+v", got) } } + +// Issue 1126 (ADR 0547 §2, §3): with the node's own compositions, unsigned, and wants naming the build alone, the +// composition composed for the want's build is applied, and only when it differs from what runs; for a module of +// policy record a move that recreates a container waits in phase waiting-yes with its judgement until the want names +// the yes for that composition, and is then applied. +func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(t *testing.T) { + b := newBed(t) + b.env.SelfComposed, b.env.Trusted = true, nil + b.env.Policy = func(m string) string { + if m == "db" { + return catalogue.PolicyRecord + } + return catalogue.PolicyRoll + } + stop := b.run() + defer stop() + unsigned := func(module, fp string, rs ...json.RawMessage) (string, []byte, string) { + c := delivery.Composed{Node: node, Module: module, FP: fp, Resources: rs} + key, _ := delivery.ComposedKey(node, module, fp) + raw, err := delivery.EncodeComposed(c) + if err != nil { + t.Fatal(err) + } + dig, _ := delivery.ComposedDigestOf(c) + return key, raw, dig + } + want := func(module, fp, yes string) { + key, _ := delivery.WantKey(node, module) + raw, err := delivery.EncodeWant(delivery.Want{FP: fp, Yes: yes}) + if err != nil { + t.Fatal(err) + } + b.wants.put(key, raw) + } + + fp := fpOf("a") + ck, craw, dig := unsigned("web", fp, file("web", "x", "1")) + b.comps.put(ck, craw) + want("web", fp, "") + eventually(t, "the node's own composition was not applied", func() bool { return len(b.appliedList()) == 1 }) + eventually(t, "the have does not name the composed digest", func() bool { return b.haves.last("web").Composed == dig }) + // The same composition handed again is no move. + b.comps.put(ck, craw) + time.Sleep(50 * time.Millisecond) + // A revision at the same build (a config value changed) is applied. + _, craw2, dig2 := unsigned("web", fp, file("web", "x", "2")) + b.comps.put(ck, craw2) + eventually(t, "the revision was not applied", func() bool { return len(b.appliedList()) == 2 && b.haves.last("web").Composed == dig2 }) + + // A record module: its first composition is no move; a new image recreates its container and waits for a yes. + container := func(image string) json.RawMessage { + raw, _ := json.Marshal(map[string]any{"id": "db.server", "type": "container", "module": "db", "name": "db", "image": image}) + return raw + } + dk, draw, _ := unsigned("db", fp, container("db@sha256:"+strings.Repeat("1", 64))) + b.comps.put(dk, draw) + want("db", fp, "") + eventually(t, "the record module's first composition was not applied", func() bool { return len(b.appliedList()) == 3 }) + fpB := fpOf("b") + dk2, draw2, ddig2 := unsigned("db", fpB, container("db@sha256:"+strings.Repeat("2", 64))) + b.comps.put(dk2, draw2) + want("db", fpB, "") + eventually(t, "the move did not wait for a yes", func() bool { + h := b.haves.last("db") + return h.Phase == string(delivery.PhaseWaitingYes) && h.Judgement == string(delivery.JudgedRecreates) && h.Waits == ddig2 + }) + if got := b.appliedList(); len(got) != 3 { + t.Fatalf("a move waiting for a yes was applied: %v", got) + } + want("db", fpB, ddig2) + eventually(t, "the yes did not apply the move", func() bool { + h := b.haves.last("db") + return len(b.appliedList()) == 4 && h.Phase == string(delivery.PhaseRunning) && h.FP == fpB + }) +} diff --git a/internal/link/have.go b/internal/link/have.go index c994c16..64f803c 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -324,6 +324,9 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { f := s.modules[module] h := delivery.Have{V: delivery.Version, FP: f.FP, Composed: f.Composed, Build: f.Build, Phase: delivery.PhaseRunning, Health: delivery.HealthUnknown, Detail: f.Detail} + if delivery.Phase(f.Phase) == delivery.PhaseWaitingYes { + h.Judgement, h.Waits = delivery.Judgement(f.Judgement), f.Waits + } switch { case f.Phase != "": h.Phase = delivery.Phase(f.Phase) diff --git a/internal/link/messages.go b/internal/link/messages.go index 3aaba65..305ac0d 100644 --- a/internal/link/messages.go +++ b/internal/link/messages.go @@ -743,4 +743,7 @@ type ModuleFact struct { Phase string Failed bool Detail string + // Judgement and Waits are, in phase waiting-yes, the judgement of the move and the composition waiting (ADR 0547 §3). + Judgement string + Waits string } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go new file mode 100644 index 0000000..42f5aa0 --- /dev/null +++ b/internal/selfcompose/selfcompose.go @@ -0,0 +1,494 @@ +// Package selfcompose is the node agent's own composer (novox/hq ADR 0547 §2, issue 1126): it composes this node's +// modules and its own composition with mesh-sdk's composer, from what is on the bus, and hands the compositions to the +// agent as the agent once read them from the controller's bucket `composed-`. +// +// **It watches, it never polls** (ADR 0547 §2): its inputs are watches of the node's wants, the indexer's placement and +// modules, every node's own have key (where a node's record rides, ADR 0555 §3) and mesh-config's resolved values of this +// node. A change of any of them marks the node to compose again; one loop drains the mark, so a burst of changes composes +// the node once (composing one node takes 0.8 to 4.4 s, measured 2026-10-26). It hands the agent only the compositions that +// differ from the ones it handed before, and takes back one no want names any more. A composition that cannot be made is +// said, once per reason, and the ones handed before stand: a node keeps running what it ran rather than nothing. +package selfcompose + +import ( + "context" + "encoding/json" + "fmt" + "sort" + "strings" + "sync" + + "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/agent" +) + +// ResolvedBucket is mesh-config's state of resolved values, keyed `..` (mesh-catalog modules/mesh-config +// resolve.go ResolvedKey); a node reads its own slice alone (ADR 0547 §2). +const ResolvedBucket = "mesh-config_resolved" + +// Env is what the composer needs of the node. +type Env struct { + Node string + // Watch follows one bucket's keys matching pattern, as the agent's watches do. + Watch func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) + // Get reads one key's latest value; nil and no error when there is none. + Get func(ctx context.Context, bucket, key string) ([]byte, error) + Mesh frombus.Mesh + Say func(string) +} + +// Composer composes the node and is the agent's watcher of its compositions (agent.Env.Composed). +type Composer struct { + env Env + + mu sync.Mutex + wants map[string]delivery.Want // by module, the node's own under agent.NodeModule + placements map[string]delivery.Placement + records map[string]delivery.NodeRecord + modules map[string]delivery.Module + values map[string]map[string]any // by module, this node's + manifests map[string][]byte // by digest, read once + handed map[string]string // composed key to the digest handed + said string + caught bool + rev uint64 +} + +// New is a composer for one node. +func New(env Env) *Composer { + if env.Say == nil { + env.Say = func(string) {} + } + return &Composer{env: env, wants: map[string]delivery.Want{}, placements: map[string]delivery.Placement{}, + records: map[string]delivery.NodeRecord{}, modules: map[string]delivery.Module{}, values: map[string]map[string]any{}, + manifests: map[string][]byte{}, handed: map[string]string{}} +} + +// input is one watched bucket and how its updates are taken. +type input struct { + bucket, pattern string + take func(c *Composer, u agent.Update) +} + +func (c *Composer) inputs() []input { + node := c.env.Node + return []input{ + {delivery.AssignerWantsBucket, "want." + node + ".>", (*Composer).takeWant}, + {delivery.PlacementBucket, ">", (*Composer).takePlacement}, + {delivery.HaveBucket, "*." + delivery.HaveNodeWord, (*Composer).takeRecord}, + {delivery.ModulesBucket, ">", (*Composer).takeModule}, + {ResolvedBucket, node + ".>", (*Composer).takeValue}, + } +} + +// Watch starts the composer and answers the compositions it makes: each as a key of `composed-` would be, then a +// Caught update once every input was read and the node composed once, then every change. It ends when ctx ends or an +// input's watch ends. +func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, error) { + ins := c.inputs() + type tagged struct { + in int + u agent.Update + ok bool + } + merged := make(chan tagged, 64) + for i, in := range ins { + ch, err := c.env.Watch(ctx, in.bucket, in.pattern) + if err != nil { + return nil, fmt.Errorf("the node composes from %s, which cannot be watched: %w", in.bucket, err) + } + go func(i int, ch <-chan agent.Update) { + for u := range ch { + select { + case merged <- tagged{in: i, u: u, ok: true}: + case <-ctx.Done(): + return + } + } + select { + case merged <- tagged{in: i}: + case <-ctx.Done(): + } + }(i, ch) + } + out := make(chan agent.Update, 64) + go func() { + defer close(out) + caught := make([]bool, len(ins)) + ready := false + dirty := false + for { + // Drain whatever came, then compose once for all of it. + select { + case <-ctx.Done(): + return + case t := <-merged: + if !t.ok { + c.env.Say(fmt.Sprintf("the watch of %s ended; the node composes no more until its agent starts again", ins[t.in].bucket)) + return + } + if t.u.Caught { + caught[t.in] = true + } else { + ins[t.in].take(c, t.u) + dirty = true + } + if len(merged) > 0 { + continue + } + } + if !ready { + ready = all(caught) + if !ready { + continue + } + dirty = true + } + if !dirty { + continue + } + dirty = false + for _, u := range c.compose(ctx) { + select { + case out <- u: + case <-ctx.Done(): + return + } + } + if !c.caughtSent() { + select { + case out <- agent.Update{Caught: true}: + case <-ctx.Done(): + return + } + } + } + }() + return out, nil +} + +func all(b []bool) bool { + for _, v := range b { + if !v { + return false + } + } + return true +} + +// caughtSent says the Caught update went out, marking it sent the first time it is asked. +func (c *Composer) caughtSent() bool { + c.mu.Lock() + defer c.mu.Unlock() + sent := c.caught + c.caught = true + return sent +} + +func (c *Composer) takeWant(u agent.Update) { + c.mu.Lock() + defer c.mu.Unlock() + if _, err := delivery.ParseWantNodeKey(u.Key); err == nil { + if u.Deleted { + delete(c.wants, agent.NodeModule) + } else { + c.wants[agent.NodeModule] = delivery.Want{} + } + return + } + node, module, err := delivery.ParseWantKey(u.Key) + if err != nil || node != c.env.Node { + return + } + w, err := delivery.DecodeWant(u.Value) + if u.Deleted || err != nil || w.Removed || w.FP == "" { + delete(c.wants, module) + return + } + c.wants[module] = w +} + +func (c *Composer) takePlacement(u agent.Update) { + node, err := delivery.ParsePlacementKey(u.Key) + if err != nil { + return + } + c.mu.Lock() + defer c.mu.Unlock() + p, err := delivery.DecodePlacement(u.Value) + if u.Deleted || err != nil || p.Node != node { + delete(c.placements, node) + return + } + c.placements[node] = p +} + +func (c *Composer) takeRecord(u agent.Update) { + node, _, isNode, err := delivery.ParseHaveKey(u.Key) + if err != nil || !isNode { + return + } + c.mu.Lock() + defer c.mu.Unlock() + h, err := delivery.DecodeHaveNode(u.Value) + if u.Deleted || err != nil || h.Record == nil { + delete(c.records, node) + return + } + c.records[node] = *h.Record +} + +func (c *Composer) takeModule(u agent.Update) { + name, err := delivery.ParseModulesKey(u.Key) + if err != nil { + return + } + c.mu.Lock() + defer c.mu.Unlock() + m, err := delivery.DecodeModule(u.Value) + if u.Deleted || err != nil { + delete(c.modules, name) + return + } + c.modules[name] = m +} + +func (c *Composer) takeValue(u agent.Update) { + rest, ok := strings.CutPrefix(u.Key, c.env.Node+".") + if !ok { + return + } + module, key, ok := strings.Cut(rest, ".") + if !ok { + return + } + c.mu.Lock() + defer c.mu.Unlock() + var r struct { + Value json.RawMessage `json:"value"` + } + var v any + if u.Deleted || json.Unmarshal(u.Value, &r) != nil || len(r.Value) == 0 || json.Unmarshal(r.Value, &v) != nil { + delete(c.values[module], key) + return + } + if c.values[module] == nil { + c.values[module] = map[string]any{} + } + c.values[module][key] = v +} + +// compose composes the node once and answers the updates to hand: each composition whose digest differs from the one +// handed, and a deletion for each handed one no want names any more. +func (c *Composer) compose(ctx context.Context) []agent.Update { + c.mu.Lock() + fps := map[string]string{} + kept := map[string]string{} + var behind []string // modules wanted at another build than the trunk's: composed with that build's manifest + for module, w := range c.wants { + if module == agent.NodeModule { + continue + } + fps[module] = w.FP + if m, ok := c.modules[module]; ok && m.FP2 != w.FP { + behind = append(behind, module) + } + } + _, nodeWanted := c.wants[agent.NodeModule] + r := reader{c: c} + c.mu.Unlock() + sort.Strings(behind) + for _, module := range behind { + digest, err := c.builtManifest(ctx, module, fps[module]) + if err != nil { + c.sayOnce(fmt.Sprintf("%s is wanted at %s, whose manifest cannot be read, so the node does not compose: %v", module, fps[module], err)) + return nil + } + kept[module] = digest + } + + compositions, _, err := frombus.Compose(ctx, r, c.env.Mesh, frombus.Scope{Node: c.env.Node, Kept: kept}, fps) + if err != nil { + c.sayOnce(fmt.Sprintf("the node cannot compose its modules, and keeps running what it runs: %v", err)) + return nil + } + c.sayOnce("") + c.mu.Lock() + defer c.mu.Unlock() + var out []agent.Update + now := map[string]bool{} + keys := make([]string, 0, len(compositions)) + for key := range compositions { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + comp := compositions[key] + if comp.Module == "" && !nodeWanted { + continue + } + now[key] = true + dig, err := delivery.ComposedDigestOf(comp) + if err != nil { + c.env.Say(fmt.Sprintf("the composition %s has no digest, and is not handed: %v", key, err)) + continue + } + if c.handed[key] == dig { + continue + } + raw, err := delivery.EncodeComposed(comp) + if err != nil { + c.env.Say(fmt.Sprintf("the composition %s cannot be written, and is not handed: %v", key, err)) + continue + } + c.rev++ + c.handed[key] = dig + out = append(out, agent.Update{Key: key, Value: raw, Revision: c.rev}) + } + for key := range c.handed { + if !now[key] { + c.rev++ + delete(c.handed, key) + out = append(out, agent.Update{Key: key, Deleted: true, Revision: c.rev}) + } + } + return out +} + +func (c *Composer) sayOnce(line string) { + c.mu.Lock() + if c.said == line { + c.mu.Unlock() + return + } + c.said = line + c.mu.Unlock() + if line != "" { + c.env.Say(line) + } +} + +// builtManifest is the digest of the manifest a module's build at a fingerprint was made from (the builder's build key): +// what a node composes a module from when its want names another build than the trunk's. +func (c *Composer) builtManifest(ctx context.Context, module, fp string) (string, error) { + key, err := delivery.BuilderBuildKey(module, fp) + if err != nil { + return "", err + } + raw, err := c.env.Get(ctx, delivery.BuilderBuildsBucket, key) + if err != nil { + return "", err + } + if raw == nil { + return "", fmt.Errorf("the builder holds no build %s", key) + } + b, err := delivery.DecodeBuilderBuild(raw) + if err != nil { + return "", err + } + if b.Manifest == "" { + return "", fmt.Errorf("the build %s names no manifest", key) + } + return b.Manifest, nil +} + +// Policy is a module's upgrade policy as its manifest at the indexer's trunk derives it (catalogue.DerivedUpgrade), for +// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read. +func (c *Composer) Policy(module string) string { + c.mu.Lock() + m, ok := c.modules[module] + c.mu.Unlock() + if !ok || m.Manifest == "" { + return "" + } + raw, err := reader{c: c}.Manifest(context.Background(), m.Manifest) + if err != nil { + return "" + } + parsed, err := catalogue.ParseManifestForInputs(raw) + if err != nil { + return "" + } + policy, _, _ := catalogue.DerivedUpgrade(parsed) + return policy +} + +// reader is the composer's view of the bus as frombus reads it. +type reader struct{ c *Composer } + +func (r reader) Placements(context.Context) (map[string]delivery.Placement, error) { + r.c.mu.Lock() + defer r.c.mu.Unlock() + out := make(map[string]delivery.Placement, len(r.c.placements)) + for k, v := range r.c.placements { + out[k] = v + } + return out, nil +} + +func (r reader) Records(context.Context) (map[string]delivery.NodeRecord, error) { + r.c.mu.Lock() + defer r.c.mu.Unlock() + out := make(map[string]delivery.NodeRecord, len(r.c.records)) + for k, v := range r.c.records { + out[k] = v + } + return out, nil +} + +func (r reader) Modules(context.Context) (map[string]delivery.Module, error) { + r.c.mu.Lock() + defer r.c.mu.Unlock() + out := make(map[string]delivery.Module, len(r.c.modules)) + for k, v := range r.c.modules { + out[k] = v + } + return out, nil +} + +func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { + r.c.mu.Lock() + raw, ok := r.c.manifests[digest] + r.c.mu.Unlock() + if ok { + return raw, nil + } + key, err := delivery.ManifestsKey(digest) + if err != nil { + return nil, err + } + value, err := r.c.env.Get(ctx, delivery.ManifestsBucket, key) + if err != nil { + return nil, err + } + if value == nil { + return nil, fmt.Errorf("the indexer holds no manifest %s", digest) + } + m, err := delivery.DecodeManifest(value) + if err != nil { + return nil, err + } + r.c.mu.Lock() + r.c.manifests[digest] = []byte(m.JSON) + r.c.mu.Unlock() + return []byte(m.JSON), nil +} + +func (r reader) Values(_ context.Context, node, module string) (map[string]any, error) { + if node != r.c.env.Node { + return nil, nil // a node reads its own slice alone (ADR 0547 §2) + } + r.c.mu.Lock() + defer r.c.mu.Unlock() + if len(r.c.values[module]) == 0 { + return nil, nil + } + out := make(map[string]any, len(r.c.values[module])) + for k, v := range r.c.values[module] { + out[k] = v + } + return out, nil +} diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go new file mode 100644 index 0000000..35c87bf --- /dev/null +++ b/internal/selfcompose/selfcompose_test.go @@ -0,0 +1,159 @@ +package selfcompose + +import ( + "context" + "strings" + "sync" + "testing" + "time" + + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/agent" +) + +const lemurs = `{"module":"lemurs","version":"1","capabilities":["systemd"],"resources":[{"id":"service","type":"service","unit":"lemurs.service","state":"running","boot":"enabled"}]}` + +// fakeBus is the buckets the composer watches, each a channel a test writes updates to, and the keys Get reads. +type fakeBus struct { + mu sync.Mutex + chans map[string]chan agent.Update + kv map[string][]byte // bucket/key + said []string +} + +func newFakeBus() *fakeBus { + return &fakeBus{chans: map[string]chan agent.Update{}, kv: map[string][]byte{}} +} + +func (b *fakeBus) ch(bucket string) chan agent.Update { + b.mu.Lock() + defer b.mu.Unlock() + if b.chans[bucket] == nil { + b.chans[bucket] = make(chan agent.Update, 16) + } + return b.chans[bucket] +} + +func (b *fakeBus) env() Env { + return Env{Node: "n1", + Watch: func(_ context.Context, bucket, _ string) (<-chan agent.Update, error) { return b.ch(bucket), nil }, + Get: func(_ context.Context, bucket, key string) ([]byte, error) { + b.mu.Lock() + defer b.mu.Unlock() + return b.kv[bucket+"/"+key], nil + }, + Say: func(s string) { b.mu.Lock(); b.said = append(b.said, s); b.mu.Unlock() }} +} + +func (b *fakeBus) put(t *testing.T, bucket, key string, value []byte) { + t.Helper() + b.ch(bucket) <- agent.Update{Key: key, Value: value} +} + +func next(t *testing.T, out <-chan agent.Update) agent.Update { + t.Helper() + select { + case u := <-out: + return u + case <-time.After(5 * time.Second): + t.Fatal("no update from the composer") + return agent.Update{} + } +} + +func quiet(t *testing.T, out <-chan agent.Update) { + t.Helper() + select { + case u := <-out: + t.Fatalf("the composer handed %s, and nothing changed", u.Key) + case <-time.After(50 * time.Millisecond): + } +} + +// Issue 1126 (ADR 0547 §2): the node composes its wanted modules and its own composition from its inputs on the bus once +// every input was read, hands each, then Caught; an input that changes nothing composed hands nothing; a module no want +// names any more is taken back; a node that cannot compose says so once and hands nothing. +func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { + b := newFakeBus() + fp := delivery.SourcePrefix2 + strings.Repeat("a", 64) + digest := delivery.ManifestDigest([]byte(lemurs)) + man, _ := delivery.EncodeManifest(delivery.Manifest{Module: "lemurs", JSON: lemurs}) + key, _ := delivery.ManifestsKey(digest) + b.kv[delivery.ManifestsBucket+"/"+key] = man + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + c := New(b.env()) + out, err := c.Watch(ctx, "n1.>") + if err != nil { + t.Fatal(err) + } + place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, + Modules: []delivery.PlacedModule{{Module: "lemurs"}}}) + have, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", Capabilities: []string{"systemd"}}}) + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}) + want, _ := delivery.EncodeWant(delivery.Want{FP: fp}) + nodeWant, _ := delivery.EncodeNodeWant(delivery.NodeWant{}) + b.put(t, delivery.PlacementBucket, "placement.n1", place) + b.put(t, delivery.HaveBucket, "n1.node", have) + b.put(t, delivery.ModulesBucket, "lemurs", mod) + b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", want) + b.put(t, delivery.AssignerWantsBucket, "want.n1.node", nodeWant) + for _, bucket := range []string{delivery.AssignerWantsBucket, delivery.PlacementBucket, delivery.HaveBucket, delivery.ModulesBucket, ResolvedBucket} { + b.ch(bucket) <- agent.Update{Caught: true} + } + got := map[string]bool{} + for { + u := next(t, out) + if u.Caught { + break + } + c, err := delivery.DecodeComposed(u.Value) + if err != nil { + t.Fatalf("%s: %v", u.Key, err) + } + if c.Node != "n1" { + t.Errorf("%s composed for %q", u.Key, c.Node) + } + got[u.Key] = true + } + lemursKey, _ := delivery.ComposedKey("n1", "lemurs", fp) + nodeKey, _ := delivery.ComposedNodeKey("n1") + if !got[lemursKey] || !got[nodeKey] || len(got) != 2 { + b.mu.Lock() + t.Logf("said: %v", b.said) + b.mu.Unlock() + t.Fatalf("handed %v, want %s and %s", got, lemursKey, nodeKey) + } + + // A config value of another module, and a value changing nothing composed, hand nothing. + b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1}`)) + quiet(t, out) + + // A want removed takes its composition back. + removed, _ := delivery.EncodeWant(delivery.Want{Removed: true}) + b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", removed) + if u := next(t, out); u.Key != lemursKey || !u.Deleted { + t.Fatalf("a removed want handed %+v", u) + } + + // A want of a module whose manifest cannot be read: said once, nothing handed, what was handed stands. + gone := delivery.SourcePrefix2 + strings.Repeat("b", 64) + missing, _ := delivery.EncodeWant(delivery.Want{FP: gone}) + b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing) + quiet(t, out) + b.put(t, delivery.AssignerWantsBucket, "want.n1.lemurs", missing) + quiet(t, out) + b.mu.Lock() + n := 0 + for _, s := range b.said { + if strings.Contains(s, "lemurs is wanted at") { + n++ + } + } + b.mu.Unlock() + if n != 1 { + t.Errorf("the fault was said %d times, want once: %v", n, b.said) + } +}