diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 44d250e..8296b75 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -56,6 +56,14 @@ func (w bucketWatch) Watch(ctx context.Context, pattern string) (<-chan agent.Up type sessionHaves struct{ session *link.Session } func (h sessionHaves) Read(ctx context.Context, module string) (agent.Have, bool) { + if module == agent.NodeModule { + // The node's own key: the digest of its own composition it runs, and what that carries sealed. + got, ok := h.session.ReadNodeHave(ctx) + if !ok || got.Composed == "" { + return agent.Have{}, false + } + return agent.Have{Module: module, Composed: got.Composed, Sealed: got.Sealed}, true + } got, ok := h.session.ReadHave(ctx, module) if !ok { return agent.Have{}, false @@ -68,7 +76,7 @@ func (h sessionHaves) Write(ctx context.Context, x agent.Have) { if x.Module == agent.NodeModule { // The node's own key holds its facts and the digest of the composition it runs. if x.Phase == "" || x.Phase == string(delivery.PhaseRunning) { - h.session.NoteNode(ctx, x.Composed) + h.session.NoteNode(ctx, x.Composed, x.Sealed) } return } @@ -511,6 +519,10 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess unseal: sealOpener(opts.state), // And what each module's have says it runs sealed: after a restart the agent has applied nothing yet. held: func(module string) []string { + if module == agent.NodeModule { + got, _ := session.ReadNodeHave(linked) + return got.Sealed + } got, ok := session.ReadHave(linked, module) if !ok { return nil @@ -523,6 +535,15 @@ func nodeAgent(ctx context.Context, opts options, signer ed25519.PublicKey, sess } }, } + // And its answer to every placement candidate, placed or not, applying nothing (novox/hq ADR 0559 §4): written into + // its own key, `.node`, under candidates. + shadowHooks := hooks + shadowHooks.refused = nil + shadow := selfcompose.NewShadow(composerEnv(session, node, say, shadowHooks), + func(ctx context.Context, answers map[string]delivery.CandidateAnswer) { + session.NoteCandidates(ctx, answers) + }) + go shadow.Keep(linked, link.Backoff) if composer := selfComposer(linked, session, node, say, hooks); composer != nil { env.Composed, env.SelfComposed, env.Trusted, env.Policy = composer, true, nil, composer.Policy } else { @@ -838,7 +859,13 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f return nil } say("this node composes its own modules: the catalogue places it (" + key + ")") - return selfcompose.New(selfcompose.Env{Node: node, + return selfcompose.New(composerEnv(session, node, say, hooks)) +} + +// composerEnv is what the node's composers read of the bus, the vault and the agent: its own composer and its answer +// to placement candidates (selfcompose.Shadow) alike. +func composerEnv(session *link.Session, node string, say func(string), hooks composerHooks) selfcompose.Env { + return selfcompose.Env{Node: node, Watch: func(ctx context.Context, bucket, pattern string) (<-chan agent.Update, error) { return bucketWatch{session, bucket}.Watch(ctx, pattern) }, @@ -857,7 +884,7 @@ func selfComposer(ctx context.Context, session *link.Session, node string, say f }, // And the vault's word that a secret changed, on which it composes again (ADR 0547 §2). Hear: session.Hear, - Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held}) + Running: hooks.running, Unseal: hooks.unseal, Refused: hooks.refused, Held: hooks.held} } // endWhenPlaced calls stop once this node's placement key appears in the indexer's placement (a value after the watch diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index ffc9813..06bc6c8 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -113,7 +113,7 @@ func (b *itBed) start(trustOperator bool) (stop func()) { } } b.session = &link.Session{Membership: link.Membership{Node: itNode}, Timeout: 2 * time.Second, - Say: b.say} + Say: b.say, Keys: func() link.RecordKeys { return link.RecordKeys{} }} b.session.Attach(context.Background(), link.OverNATS{Conn: b.conn, JS: b.js, Node: itNode}) ctx, cancel := context.WithCancel(context.Background()) done := make(chan struct{}) @@ -316,7 +316,8 @@ func TestNatsTheHandPathAppliesACompositionTheOperatorSigned(t *testing.T) { // Issue 1126 (ADR 0547 §2, the milestone's H8): the node agent started whole against a real bus, once the catalogue // places its node (the placement written after it started), composes a module from its want, the indexer's manifest, its record and its mesh-config value, and applies it, // with no key in the controller's compositions; a changed config value is composed again on its event and applied as a -// revision of what runs. +// revision of what runs. And a catalogue pull request's placement candidate equal to its placement is answered "same" +// in the node's own key, and nothing is applied for it (ADR 0559 §4, subtask 10 of the milestone). func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { b := newItBed(t) kvOf := func(bucket string) nats.KeyValue { @@ -329,6 +330,11 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { placement, modules, manifests := kvOf(delivery.PlacementBucket), kvOf(delivery.ModulesBucket), kvOf(delivery.ManifestsBucket) resolved := kvOf(selfcompose.ResolvedBucket) kvOf(delivery.BuilderBuildsBucket) + candidates := kvOf(delivery.PlacementCandidatesBucket) + // The controller's stream the node's facts go into, so the node's key carries the facts it says. + if _, err := b.js.AddStream(&nats.StreamConfig{Name: "CONTROL", Subjects: []string{link.ReportSubject(itNode)}}); err != nil { + t.Fatal(err) + } path := filepath.Join(b.dir, "web.conf") man := `{"module":"web","version":"1","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` + `"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0644","content":"greeting=${config:greeting}\n"}]}` @@ -385,4 +391,37 @@ func TestNatsANodeThePlacementNamesComposesAndAppliesItsOwn(t *testing.T) { h := b.have("web") return h.FP == fp && h.Composed != first && h.Phase == delivery.PhaseRunning }) + + // The node says its facts (its record, as the test wrote it), then a candidate equal to its placement is written: the + // node answers "same" at the candidate's head and revision in its own key, and applies nothing. + b.session.SayFacts(context.Background(), link.Report{Profile: map[string]any{"capabilities": []any{ + map[string]any{"name": "service-manager", "present": true}}}}) + second := b.have("web").Composed + same, _ := delivery.DecodePlacement(place) + craw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: "feat/x", + Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h1"}, Placement: &same}) + if err != nil { + t.Fatal(err) + } + ckey, _ := delivery.PlacementCandidateKey("feat/x", itNode) + crev, err := candidates.Put(ckey, craw) + if err != nil { + t.Fatal(err) + } + var answer delivery.CandidateAnswer + within(t, 10*time.Second, "the node did not answer its candidate", func() bool { + e, err := b.haves.Get(hkey) + if err != nil { + return false + } + n, err := delivery.DecodeHaveNode(e.Value()) + answer = n.Candidates["feat/x"] + return err == nil && answer.Head == "h1" && n.Record != nil + }) + if answer.Verdict != delivery.CandidateSame || answer.Revision != crev || len(answer.Same) != 1 || answer.Same[0] != "web" { + t.Errorf("the answer to a candidate equal to the placement: %+v", answer) + } + if h := b.have("web"); h.Composed != second || b.file("web") != "greeting=salut\n" { + t.Errorf("a candidate moved what runs: %+v, %q", h, b.file("web")) + } } diff --git a/go.mod b/go.mod index be6daa0..0a71ab3 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.20261026210326-979644420117 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f 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 e8660dd..04ad7e0 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,15 @@ git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa h1:tT6sl3vY0iVAXw+mqy3iGKUBUjVBRWvQvKUOTmTyWwk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011211-82f10c3bc9fa/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef h1:uMqybf/sTAjapyllVjy8mjCi8evpcOmgnjmqRdJRsFo= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027011927-ba398e2460ef/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56 h1:PUc5s3oaWEg0uyVmXhJSSK3x0IXZBXWGqq30U/CSx30= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012259-a37d2622ed56/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f h1:AKzV973UTWe0OYFoIjgGu0f+B26xh8QsAHzCpsTIjM8= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027012704-062c29c3a22f/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f h1:mimAGERguQmCGYVNdRTrFcZ1dhpU6GWo+Q3EDhHOrVk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261027014310-cadd7b1b2f8f/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 c49c539..dd101be 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -135,7 +135,7 @@ type Env struct { // 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 + Policy func(ctx context.Context, module string) string } // transientFor is how many transient retries say the phase applying before the have says failed: past them the bus has @@ -377,8 +377,8 @@ func (a *Agent) NotComposed(ctx context.Context, why string) { } } -// writeHave writes a module's have; on a node composing its own, with the sealed values of the composition it names -// when the caller gave none (Have.Sealed): those of the composition the agent applied or holds at that digest, else +// writeHave writes a module's have (the node's own key for NodeModule); on a node composing its own, with the sealed +// values of the composition it names when the caller gave none (Have.Sealed): those of the composition the agent applied or holds at that digest, else // those the module's last have said at that digest. // // A writer that names no composition (one written from the agent's running entry before this agent ran anything of the @@ -394,7 +394,9 @@ func (a *Agent) writeHave(ctx context.Context, h Have) { } } } - if a.env.SelfComposed && h.Sealed == nil && h.Module != NodeModule && h.Composed != "" { + // The node's own composition too: its core boxes are sealed again by the vault at a restart, and kept in its key the + // composer reuses them, so the restart applies nothing again (review of L2, round 3, finding 1). + if a.env.SelfComposed && h.Sealed == nil && h.Composed != "" { h.Sealed = a.sealedOf(ctx, h.Module, h.FP, h.Composed) } a.env.Haves.Write(ctx, h) @@ -1101,7 +1103,7 @@ func short(s string) string { // 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 { + if module == NodeModule || a.env.Policy == nil || a.env.Policy(ctx, module) != catalogue.PolicyRecord || w.Want.Yes == comp.Dig { return false } a.mu.Lock() diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index 44290f5..ed10ab9 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -964,7 +964,7 @@ func TestAModuleRunningFromTheKeptDeclarationKeepsItsFilledPaths(t *testing.T) { func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes(t *testing.T) { b := newBed(t) b.env.SelfComposed, b.env.Trusted = true, nil - b.env.Policy = func(m string) string { + b.env.Policy = func(_ context.Context, m string) string { if m == "db" || m == "rec" { return catalogue.PolicyRecord } @@ -1001,6 +1001,21 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( // The same composition handed again is no move. b.comps.put(ck, craw) time.Sleep(50 * time.Millisecond) + // The node's own composition carries its sealed values in its key, as a module's have does, so a restart reuses its + // core boxes and applies nothing again (review of L2, round 3, finding 1). + box, _ := json.Marshal(map[string]any{"id": "core-box", "type": "file", "path": "/tmp/r6/core-box", "sealed": "core"}) + own := delivery.Composed{Node: node, Resources: []json.RawMessage{box}} + nk, _ := delivery.ComposedNodeKey(node) + nraw, _ := delivery.EncodeComposed(own) + ndig, _ := delivery.ComposedDigestOf(own) + b.comps.put(nk, nraw) + nwk, _ := delivery.WantNodeKey(node) + nwraw, _ := delivery.EncodeNodeWant(delivery.NodeWant{Self: true}) + b.wants.put(nwk, nwraw) + eventually(t, "the node's own key does not carry its sealed values", func() bool { + h := b.haves.last(NodeModule) + return h.Composed == ndig && len(h.Sealed) == 1 && h.Sealed[0] == "core" + }) // The node cannot compose (a stand-in refused): each wanted module's have says so, failed, naming why, and still // names what runs; composed again, the have is what it runs. web was applied by this agent; idle was not since the // agent's start (the switch and every delivery of the node-engine restart it), so what runs is what its have said. @@ -1035,7 +1050,7 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( // 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 }) + eventually(t, "the revision was not applied", func() bool { return len(b.appliedList()) == 3 && 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 { @@ -1046,7 +1061,7 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( 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 }) + eventually(t, "the record module's first composition was not applied", func() bool { return len(b.appliedList()) == 4 }) fpB := fpOf("b") dk2, draw2, ddig2 := unsigned("db", fpB, container("db@sha256:"+strings.Repeat("2", 64))) // rec is a record module this agent has run nothing of since its start (the switch and every delivery of the diff --git a/internal/link/have.go b/internal/link/have.go index f2b1931..d923ee8 100644 --- a/internal/link/have.go +++ b/internal/link/have.go @@ -47,6 +47,11 @@ type haveState struct { 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 + // nodeSealed is, on a node composing its own, each value its own composition carries sealed to its key (ADR 0547 §3, + // issue 1126): kept in its key so a restart reuses them and applies nothing again. + nodeSealed []string + // nodeCandidates is the node's answer to each placement candidate group (novox/hq ADR 0559 §4), written into its key. + nodeCandidates map[string]delivery.CandidateAnswer // 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). @@ -416,6 +421,8 @@ func (q *Session) writeHaves(ctx context.Context, bus Bus) { var raw []byte facts := *s.node facts.Composed = s.nodeComposed + facts.Sealed = s.nodeSealed + facts.Candidates = s.nodeCandidates if raw, err = delivery.EncodeHaveNode(facts); err == nil { want := facts if !q.putHave(write, bucket, key, raw, func(current []byte) bool { @@ -703,12 +710,14 @@ func (q *Session) NoteRemoved(ctx context.Context, module, build string) { 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) { +// NoteNode takes the digest of the node's own composition it runs and the values it carries sealed to the node's key +// (nil for a composition the node did not make itself), and writes the node's key where it differs. +func (q *Session) NoteNode(ctx context.Context, composed string, sealed []string) { q.init() s := &q.haves s.mu.Lock() s.nodeComposed = composed + s.nodeSealed = sealed if s.node == nil { n := delivery.HaveNode{V: delivery.Version} s.node = &n @@ -720,6 +729,61 @@ func (q *Session) NoteNode(ctx context.Context, composed string) { q.writeHaves(ctx, bus) } +// NoteCandidates takes the node's answers to the placement candidates, by group (its shadow composer's, novox/hq ADR +// 0559 §4), and writes the node's key where it differs; a group left out is no longer answered. +func (q *Session) NoteCandidates(ctx context.Context, answers map[string]delivery.CandidateAnswer) { + q.init() + s := &q.haves + s.mu.Lock() + s.nodeCandidates = answers + if len(answers) == 0 { + s.nodeCandidates = nil + } + // Before the node's facts are first said, its key is not written for an answer alone: it would carry no record. The + // answers ride with the first facts. + known := s.node != nil + s.mu.Unlock() + if !known { + return + } + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + q.writeHaves(ctx, bus) +} + +// ReadNodeHave is the node's own key as the bucket holds it, and whether it holds one: what the node's own composition +// it last ran is and carries sealed, read after a restart (ADR 0547 §3). +func (q *Session) ReadNodeHave(ctx context.Context) (delivery.HaveNode, bool) { + q.init() + q.mu.Lock() + bus := q.bus + q.mu.Unlock() + hb, ok := bus.(HaveBus) + if !ok || bus == nil { + return delivery.HaveNode{}, false + } + key, err := delivery.HaveNodeKey(q.Membership.Node) + if err != nil { + return delivery.HaveNode{}, false + } + read, cancel := context.WithTimeout(ctx, q.timeout()) + defer cancel() + bucket, err := hb.HaveBind(read) + if err != nil { + return delivery.HaveNode{}, false + } + value, _, err := bucket.Get(read, key) + if err != nil { + return delivery.HaveNode{}, false + } + n, err := delivery.DecodeHaveNode(value) + if err != nil { + return delivery.HaveNode{}, false + } + return n, true +} + // ReadHave is the have this node-engine last wrote for a module, and whether the bucket holds one. A bucket that cannot // be read answers as none: the agent then compares with the declaration it kept, and applies only what differs. func (q *Session) ReadHave(ctx context.Context, module string) (delivery.Have, bool) { diff --git a/internal/link/have_test.go b/internal/link/have_test.go index fdb66ca..309fb4d 100644 --- a/internal/link/have_test.go +++ b/internal/link/have_test.go @@ -582,9 +582,22 @@ func TestAnAppliedModulesHaveNamesTheFingerprintAndTheDigestAndFollowsItsHealth( if got := l.have(t, "n1.web"); got.Phase != delivery.PhaseRemoved { t.Fatalf("removed: %+v", got) } - q.NoteNode(context.Background(), dig) - if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig { - t.Fatalf("the node key: %+v %v", n, err) + q.NoteNode(context.Background(), dig, []string{"core"}) + if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Composed != dig || len(n.Sealed) != 1 || n.Sealed[0] != "core" { + t.Fatalf("the node key, with its own composition's sealed values: %+v %v", n, err) + } + if n, ok := q.ReadNodeHave(context.Background()); !ok || n.Composed != dig || len(n.Sealed) != 1 { + t.Fatalf("the node key read back: %+v %v", n, ok) + } + // The node's answers to placement candidates ride in the same key (ADR 0559 §4), and go when no group is answered. + answers := map[string]delivery.CandidateAnswer{"feat/x": {Head: "h", Verdict: delivery.CandidateSame, Same: []string{"web"}}} + q.NoteCandidates(context.Background(), answers) + if n, err := delivery.DecodeHaveNode(l.kv["n1.node"].value); err != nil || n.Candidates["feat/x"].Head != "h" || n.Composed != dig { + t.Fatalf("the node key with its answers: %+v %v", n, err) + } + q.NoteCandidates(context.Background(), nil) + if n, _ := delivery.DecodeHaveNode(l.kv["n1.node"].value); n.Candidates != nil { + t.Fatalf("an answer left after its group went: %+v", n.Candidates) } } diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 19a4d9c..12ef2a7 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -453,11 +453,10 @@ func (c *Composer) runningBox(box string) string { } if c.env.Held != nil { c.mu.Lock() + // The node's own key among them (agent.NodeModule): its core boxes, as its last run said them. modules := make([]string, 0, len(c.wants)) for m := range c.wants { - if m != agent.NodeModule { - modules = append(modules, m) - } + modules = append(modules, m) } c.mu.Unlock() sort.Strings(modules) @@ -514,15 +513,16 @@ func (c *Composer) builtManifest(ctx context.Context, module, fp string) (string } // 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 { +// the agent's judgement of a move (ADR 0547 §3); empty when the manifest is not known or cannot be read. The manifest is +// read within ctx, the agent's (review of L2, round 1, finding 5), so a stopping agent stops the read. +func (c *Composer) Policy(ctx context.Context, 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) + raw, err := reader{c: c}.Manifest(ctx, m.Manifest) if err != nil { return "" } diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 8ade9a3..ba3a246 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -10,6 +10,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" "github.com/novox/mesh-host/internal/agent" @@ -237,14 +238,18 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } return sealed[0] } + // And the node's own key (review of L2, round 3, finding 1): a core box its own composition carries is held there, and + // kept after a restart as a module's is, so the node's own composition does not move either. for _, row := range []struct { - name, held string - kept bool - }{{"the have holds the rotated value", boxOf(string(rotated.Value)), true}, {"the have holds the value before the rotation", boxOf(before), false}} { + name, held, in string + kept bool + }{{"the have holds the rotated value", boxOf(string(rotated.Value)), "keeper", true}, + {"the have holds the value before the rotation", boxOf(before), "keeper", false}, + {"the node's own key holds the rotated value", boxOf(string(rotated.Value)), agent.NodeModule, true}} { restarted := env restarted.Running = func() []delivery.Composed { return nil } restarted.Held = func(module string) []string { - if module == "keeper" { + if module == row.in { return []string{row.held} } return nil @@ -258,7 +263,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { return nil, fmt.Errorf("not sealed to n1: %s", box) } again := New(restarted) - again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}} + again.wants = map[string]delivery.Want{"keeper": {FP: fp, Self: true}, agent.NodeModule: {}} rows, err := reader{c: again}.OwnSecrets(ctx, "n1") if err != nil || len(rows) != 1 { t.Fatalf("%s: own secrets %+v, %v", row.name, rows, err) @@ -313,4 +318,292 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { if asked, _ := json.Marshal(sent["credentials"]); !strings.Contains(string(asked), `"role":"consumer"`) || !strings.Contains(string(asked), `"role":"provider"`) { t.Errorf("the vault was asked %s", asked) } + // ADR 0559 §4: the node answers placement candidates in shadow, placed or not, applying nothing and handing no + // composition: the same; a module the group adds, its secret not yet issued; a module removed; a placement it cannot + // compose; a group whose keys went is no longer answered. What runs is never composed from a candidate: the + // Composer does not read the state. + for _, in := range New(env).inputs() { + if in.bucket == delivery.PlacementCandidatesBucket { + t.Fatal("the node's own composer reads placement candidates") + } + } + sb := newPatternBus() + trunk := delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []delivery.PlacedModule{{Module: "lemurs"}}} + mods := map[string]delivery.Module{"lemurs": {FP2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}, + "keeper": {FP2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"}} + readOnly := &fakeBus{kv: map[string][]byte{delivery.ManifestsBucket + "/" + key: man, delivery.ManifestsBucket + "/" + kkey: kman}} + senv := readOnly.env() + senv.Watch = sb.watch + senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) { + return json.RawMessage(`{"node":"n1","key":"sk","own":[]}`), nil // nothing issued for keeper: it is not placed here + } + // What n1 runs: lemurs and its own composition, as its composer composes them from the trunk. + running, _, err := frombus.Compose(ctx, fixedReader{placements: map[string]delivery.Placement{"n1": trunk}, + records: map[string]delivery.NodeRecord{"n1": {Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}, + modules: mods, manifests: map[string][]byte{digest: []byte(lemurs), kdigest: []byte(keeper)}}, frombus.Mesh{}, frombus.Scope{Node: "n1"}, map[string]string{"lemurs": fp}) + if err != nil { + t.Fatal(err) + } + senv.Running = func() []delivery.Composed { + var out []delivery.Composed + for _, c := range running { + out = append(out, c) + } + return out + } + answered := make(chan map[string]delivery.CandidateAnswer, 8) + sh := NewShadow(senv, func(_ context.Context, a map[string]delivery.CandidateAnswer) { answered <- a }) + sctx, scancel := context.WithCancel(ctx) + defer scancel() + // Review of L2b, round 1, finding 2: the candidates' state is not there when the node starts (a delivery before the + // indexer made it): the shadow tries again, and answers once it appears. + sb.missing(delivery.PlacementCandidatesBucket, true) + go sh.Keep(sctx, func(int) time.Duration { return 5 * time.Millisecond }) + time.Sleep(30 * time.Millisecond) + sb.missing(delivery.PlacementCandidatesBucket, false) + sb.made(6) + rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) + tp, _ := delivery.EncodePlacement(trunk) + sb.put(delivery.PlacementBucket, "placement.n1", tp, 1) + sb.put(delivery.HaveBucket, "n1.node", rec, 1) + for name, m := range mods { + raw, _ := delivery.EncodeModule(m) + sb.put(delivery.ModulesBucket, name, raw, 1) + } + candidate := func(group string, rev uint64, modules ...string) { + p := trunk + p.Modules = nil + for _, m := range modules { + p.Modules = append(p.Modules, delivery.PlacedModule{Module: m}) + } + raw, err := delivery.EncodePlacementCandidate(delivery.PlacementCandidate{Group: group, + Pull: delivery.Head{Repository: delivery.CatalogueRepository, Number: 1, Head: "h-" + group}, Placement: &p}) + if err != nil { + t.Fatal(err) + } + k, _ := delivery.PlacementCandidateKey(group, "n1") + sb.put(delivery.PlacementCandidatesBucket, k, raw, rev) + } + candidate("same", 3, "lemurs") + candidate("adds", 4, "lemurs", "keeper") + candidate("drops", 5) + candidate("ghost", 6, "lemurs", "ghost") + sb.caught() + var answers map[string]delivery.CandidateAnswer + waitFor := func(why string, ok func(map[string]delivery.CandidateAnswer) bool) { + t.Helper() + deadline := time.After(10 * time.Second) + for answers == nil || !ok(answers) { + select { + case answers = <-answered: + case <-deadline: + t.Fatalf("%s: last answers %+v", why, answers) + } + } + } + waitFor("every group answered", func(a map[string]delivery.CandidateAnswer) bool { return len(a) == 4 }) + for group, want := range map[string]func(delivery.CandidateAnswer) bool{ + "same": func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateSame && a.Head == "h-same" && a.Revision == 3 && len(a.Same) == 2 + }, + "adds": func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" && + len(a.Added[0].Unissued) == 1 && len(a.Refusals) == 0 + }, + "drops": func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateDiffers && len(a.Removed) == 1 && a.Removed[0] == "lemurs" + }, + "ghost": func(a delivery.CandidateAnswer) bool { + return a.Verdict == delivery.CandidateRefused && a.Refused != "" + }, + } { + if !want(answers[group]) { + t.Errorf("%s: %+v", group, answers[group]) + } + } + // The node's own key changing (its answers written) composes nothing again; a group's key deleted drops its answer. + sb.put(delivery.HaveBucket, "n1.node", rec, 2) + select { + case a := <-answered: + t.Errorf("answers handed again for nothing changed: %+v", a) + case <-time.After(100 * time.Millisecond): + } + k, _ := delivery.PlacementCandidateKey("same", "n1") + sb.del(delivery.PlacementCandidatesBucket, k) + waitFor("the deleted group's answer stays", func(a map[string]delivery.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 3 }) + + // Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches + // again. The new run reads afresh, so that group is no longer answered. + sb.missing(delivery.PlacementCandidatesBucket, true) + sb.end(delivery.ModulesBucket) + dropsKey, _ := delivery.PlacementCandidateKey("drops", "n1") + sb.del(delivery.PlacementCandidatesBucket, dropsKey) + time.Sleep(30 * time.Millisecond) + sb.missing(delivery.PlacementCandidatesBucket, false) + waitFor("a group deleted while the shadow was down is still answered", func(a map[string]delivery.CandidateAnswer) bool { + _, ok := a["drops"] + return !ok && len(a) == 2 + }) +} + +// patternBus is a bus whose every watch gets the updates of its bucket matching its pattern, as the bus's own watches +// do: two watches of one bucket each see what matches theirs. +type patternBus struct { + mu sync.Mutex + watches []patternWatch + absent map[string]bool + // kv is each bucket's keys now; once replay is set, a new watch gets those matching it and then Caught, as a watch + // of the bus's own does. + kv map[string]map[string]agent.Update + replay bool +} + +// end ends every watch of a bucket, as a watch the bus drops, and from then on each new watch replays what is there. +func (b *patternBus) end(bucket string) { + b.mu.Lock() + defer b.mu.Unlock() + b.replay = true + kept := b.watches[:0] + for _, w := range b.watches { + if w.bucket == bucket { + close(w.ch) + continue + } + kept = append(kept, w) + } + b.watches = kept +} + +// missing makes a bucket's watch fail, as one the bus holds no state for, or work again. +func (b *patternBus) missing(bucket string, gone bool) { + b.mu.Lock() + defer b.mu.Unlock() + if b.absent == nil { + b.absent = map[string]bool{} + } + b.absent[bucket] = gone +} + +type patternWatch struct { + bucket, pattern string + ch chan agent.Update +} + +func newPatternBus() *patternBus { return &patternBus{} } + +func (b *patternBus) watch(_ context.Context, bucket, pattern string) (<-chan agent.Update, error) { + b.mu.Lock() + defer b.mu.Unlock() + if b.absent[bucket] { + return nil, fmt.Errorf("nats: bucket not found") + } + w := patternWatch{bucket: bucket, pattern: pattern, ch: make(chan agent.Update, 64)} + b.watches = append(b.watches, w) + if b.replay { + for key, u := range b.kv[bucket] { + if matches(pattern, key) { + w.ch <- u + } + } + w.ch <- agent.Update{Caught: true} + } + return w.ch, nil +} + +func matches(pattern, key string) bool { + p, k := strings.Split(pattern, "."), strings.Split(key, ".") + for i, tok := range p { + if tok == ">" { + return len(k) > i + } + if i >= len(k) || (tok != "*" && tok != k[i]) { + return false + } + } + return len(p) == len(k) +} + +func (b *patternBus) send(bucket string, u agent.Update) { + b.mu.Lock() + defer b.mu.Unlock() + if !u.Caught { + if b.kv == nil { + b.kv = map[string]map[string]agent.Update{} + } + if b.kv[bucket] == nil { + b.kv[bucket] = map[string]agent.Update{} + } + if u.Deleted { + delete(b.kv[bucket], u.Key) + } else { + b.kv[bucket][u.Key] = u + } + } + for _, w := range b.watches { + if w.bucket == bucket && (u.Caught || matches(w.pattern, u.Key)) { + select { + case w.ch <- u: + default: // a watch of a run that ended reads no more + } + } + } +} + +func (b *patternBus) put(bucket, key string, value []byte, rev uint64) { + b.send(bucket, agent.Update{Key: key, Value: value, Revision: rev}) +} + +func (b *patternBus) del(bucket, key string) { b.send(bucket, agent.Update{Key: key, Deleted: true}) } + +// made waits until n watches are made, so nothing put is lost. +func (b *patternBus) made(n int) { + for { + b.mu.Lock() + got := len(b.watches) + b.mu.Unlock() + if got >= n { + return + } + time.Sleep(5 * time.Millisecond) + } +} + +// caught says every watch caught up. +func (b *patternBus) caught() { + b.mu.Lock() + defer b.mu.Unlock() + for _, w := range b.watches { + w.ch <- agent.Update{Caught: true} + } +} + +// fixedReader is frombus's Reader over values held, for composing what a node runs. +type fixedReader struct { + placements map[string]delivery.Placement + records map[string]delivery.NodeRecord + modules map[string]delivery.Module + manifests map[string][]byte +} + +func (r fixedReader) Placements(context.Context) (map[string]delivery.Placement, error) { + return r.placements, nil +} +func (r fixedReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { + return r.records, nil +} +func (r fixedReader) Modules(context.Context) (map[string]delivery.Module, error) { + return r.modules, nil +} +func (r fixedReader) Manifest(_ context.Context, d string) ([]byte, error) { + if raw, ok := r.manifests[d]; ok { + return raw, nil + } + return nil, fmt.Errorf("no manifest %s", d) +} +func (r fixedReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil } +func (r fixedReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { + return nil, nil +} +func (r fixedReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { + return nil, nil, nil } diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go new file mode 100644 index 0000000..2bbdb1d --- /dev/null +++ b/internal/selfcompose/shadow.go @@ -0,0 +1,358 @@ +package selfcompose + +import ( + "context" + "fmt" + "sort" + "sync" + "time" + + "git.novox.be/novox/mesh-sdk/go/compose/frombus" + "git.novox.be/novox/mesh-sdk/go/delivery" + + "github.com/novox/mesh-host/internal/agent" +) + +// Shadow is the node's answer to placement candidates (novox/hq ADR 0559 §4, issue 1126): for every group with a key in +// the indexer's state `placement-candidates`, it composes this node in shadow from the trunk's placement with the group's +// keys laid over it and its own inputs (the sealed values and boxes it holds among them), compares the result with what +// the node runs, and hands the answers to be written into the node's own have key, `.node`, under `candidates`. +// It applies nothing and hands no composition: what runs is still composed from `placement` alone, by the Composer. +// +// **Event-driven, like the Composer.** It watches every candidate key, the trunk's placement, every node's record, the +// indexer's modules, this node's config values, and this node's module haves (what it runs moved); a burst of changes +// answers once. It runs whether or not the node is placed yet: a node that applies the controller's compositions answers +// against what it runs from them. An answer is handed only when it changes, and a group's answer is dropped when its +// keys go. +type Shadow struct { + c *Composer + answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer) + + mu sync.Mutex + candidates map[string]map[string]candidateKey // by group, by node + answers map[string]delivery.CandidateAnswer +} + +// candidateKey is one candidate key as last read: its value and revision. +type candidateKey struct { + c delivery.PlacementCandidate + rev uint64 +} + +// NewShadow is the shadow composer of one node; answer writes the node's answers into its own have key. +func NewShadow(env Env, answer func(ctx context.Context, answers map[string]delivery.CandidateAnswer)) *Shadow { + return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{}, + answers: map[string]delivery.CandidateAnswer{}} +} + +// Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch +// yet (the candidates' state not made, or the node's grant of it not loaded at a delivery) or a watch that ended is +// tried again rather than leaving the node answering nothing for the rest of the link (review of L2b, round 1, finding +// 2). Each distinct reason is said once. +func (s *Shadow) Keep(ctx context.Context, backoff func(try int) time.Duration) { + said := "" + for try := 1; ctx.Err() == nil; try++ { + began := time.Now() + err := s.Run(ctx) + if ctx.Err() != nil { + return + } + if time.Since(began) > time.Minute { + try = 1 // it ran: a later stop begins the waits again + } + why := "it stopped" + if err != nil { + why = err.Error() + } + if why != said { + s.c.env.Say("this node answers no placement candidate for now, and tries again: " + why) + said = why + } + select { + case <-ctx.Done(): + return + case <-time.After(backoff(try)): + } + } +} + +// Run answers until ctx ends or an input's watch ends. +func (s *Shadow) Run(ctx context.Context) error { + // Each run watches afresh and starts from nothing read: what changed while it was down (a candidate key deleted) + // is then not missed, and the watches of a run before are not left open (review of L2b, round 2). + ctx, cancel := context.WithCancel(ctx) + defer cancel() + s.mu.Lock() + s.candidates = map[string]map[string]candidateKey{} + s.c = New(s.c.env) + s.mu.Unlock() + c := s.c + node := c.env.Node + ins := []input{ + {delivery.PlacementCandidatesBucket, ">", nil}, + {delivery.PlacementBucket, ">", (*Composer).takePlacement}, + {delivery.HaveBucket, "*." + delivery.HaveNodeWord, nil}, + {delivery.ModulesBucket, ">", (*Composer).takeModule}, + {ResolvedBucket, node + ".>", (*Composer).takeValue}, + // What this node runs moved: an apply ends in its module's have. + {delivery.HaveBucket, node + ".*", nil}, + } + 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 fmt.Errorf("the node answers placement candidates 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) + } + caught := make([]bool, len(ins)) + ready, dirty := false, false + for { + select { + case <-ctx.Done(): + return nil + case t := <-merged: + if !t.ok { + return fmt.Errorf("the watch of %s ended; the node answers no placement candidate until its agent starts again", ins[t.in].bucket) + } + switch { + case t.u.Caught: + caught[t.in] = true + case t.in == 0: + dirty = s.takeCandidate(t.u) || dirty + case t.in == 2: + dirty = s.takeRecord(t.u) || dirty + case t.in == 5: + dirty = !isNodeKey(t.u.Key) || dirty + default: + ins[t.in].take(c, t.u) + dirty = true + } + if len(merged) > 0 { + continue + } + } + if !ready { + if ready = all(caught); !ready { + continue + } + dirty = true + } + if dirty { + dirty = false + s.answerAll(ctx) + } + } +} + +func isNodeKey(key string) bool { + _, _, isNode, err := delivery.ParseHaveKey(key) + return err == nil && isNode +} + +// takeCandidate takes one candidate key, and says whether it changed anything. +func (s *Shadow) takeCandidate(u agent.Update) bool { + group, node, err := delivery.ParsePlacementCandidateKey(u.Key) + if err != nil { + return false + } + s.mu.Lock() + defer s.mu.Unlock() + if u.Deleted { + if _, ok := s.candidates[group][node]; !ok { + return false + } + delete(s.candidates[group], node) + if len(s.candidates[group]) == 0 { + delete(s.candidates, group) + } + return true + } + c, err := delivery.DecodePlacementCandidate(u.Value) + if err != nil || c.Group != group || (c.Placement != nil && c.Placement.Node != node) { + // Not one: read as gone. + if _, ok := s.candidates[group][node]; ok { + delete(s.candidates[group], node) + return true + } + return false + } + if s.candidates[group] == nil { + s.candidates[group] = map[string]candidateKey{} + } + s.candidates[group][node] = candidateKey{c: c, rev: u.Revision} + return true +} + +// takeRecord takes a node's record, and says whether it changed: the node's own key changes with each answer written, +// which moves nothing composed. +func (s *Shadow) takeRecord(u agent.Update) bool { + node, _, isNode, err := delivery.ParseHaveKey(u.Key) + if err != nil || !isNode { + return false + } + c := s.c + c.mu.Lock() + before, had := c.records[node] + c.mu.Unlock() + c.takeRecord(u) + c.mu.Lock() + after, has := c.records[node] + c.mu.Unlock() + return had != has || (has && !delivery.SameNodeRecord(before, after)) +} + +// answerAll answers every group and hands the answers when any changed. +func (s *Shadow) answerAll(ctx context.Context) { + c := s.c + node := c.env.Node + s.mu.Lock() + groups := make([]string, 0, len(s.candidates)) + for g := range s.candidates { + groups = append(groups, g) + } + s.mu.Unlock() + sort.Strings(groups) + next := map[string]delivery.CandidateAnswer{} + for _, group := range groups { + if a, ok := s.answerGroup(ctx, group); ok { + next[group] = a + } + } + s.mu.Lock() + same := len(next) == len(s.answers) + for g, a := range next { + if old, ok := s.answers[g]; !ok || !sameAnswer(old, a) { + same = false + } + } + if !same { + s.answers = next + } + s.mu.Unlock() + if !same && s.answer != nil { + out := make(map[string]delivery.CandidateAnswer, len(next)) + for g, a := range next { + out[g] = a + } + s.answer(ctx, out) + c.env.Say(fmt.Sprintf("%s answered %d placement candidate group(s)", node, len(out))) + } +} + +// answerGroup is the node's answer to one group, and false when the node has nothing to answer: neither the trunk nor +// the group places it. +func (s *Shadow) answerGroup(ctx context.Context, group string) (delivery.CandidateAnswer, bool) { + c := s.c + node := c.env.Node + s.mu.Lock() + keys := s.candidates[group] + given := make(map[string]*delivery.Placement, len(keys)) + var head string + for n, k := range keys { + given[n] = k.c.Placement + head = k.c.Pull.Head + } + // The revision answered is the node's own key's, 0 when the group has none for it (judged by the head alone). + own, mine := keys[node] + rev := own.rev + s.mu.Unlock() + + c.mu.Lock() + _, placed := c.placements[node] + modules := make(map[string]delivery.Module, len(c.modules)) + for k, v := range c.modules { + modules[k] = v + } + c.mu.Unlock() + if !placed && !mine { + return delivery.CandidateAnswer{}, false + } + + var running []delivery.Composed + if c.env.Running != nil { + running = c.env.Running() + } + if p, ok := given[node]; ok && p == nil { + // The head removes this node's placement file: everything it runs is removed. + a := delivery.CandidateAnswer{Head: head, Revision: rev} + for _, r := range running { + if r.Module == "" { + a.Removed = append(a.Removed, delivery.HaveNodeWord) + } else { + a.Removed = append(a.Removed, r.Module) + } + } + sort.Strings(a.Removed) + a.Verdict = delivery.VerdictOf(a) + return a, true + } + + // Each module at the build it runs; one the candidate adds at the trunk's build. + fps := map[string]string{} + kept := map[string]string{} + for _, r := range running { + if r.Module != "" && r.FP != "" { + fps[r.Module] = r.FP + } + } + placement := given[node] + if placement == nil { + c.mu.Lock() + p := c.placements[node] + c.mu.Unlock() + placement = &p + } + for _, m := range placement.Modules { + if _, runs := fps[m.Module]; !runs { + if mod, ok := modules[m.Module]; ok && mod.FP2 != "" { + fps[m.Module] = mod.FP2 + } + } + } + for module, fp := range fps { + if mod, ok := modules[module]; ok && mod.FP2 == fp { + continue + } + digest, err := c.builtManifest(ctx, module, fp) + if err != nil { + return delivery.CandidateAnswer{Head: head, Revision: rev, Verdict: delivery.CandidateRefused, + Refused: fmt.Sprintf("%s runs at %s, whose manifest cannot be read: %v", module, fp, err)}, true + } + kept[module] = digest + } + // The boxes it runs, by what each module's have and its own key say after a restart (Held), are reused for the + // modules it runs: a value sealed again is no difference. + held := map[string]delivery.Want{agent.NodeModule: {}} + for module := range fps { + held[module] = delivery.Want{FP: fps[module]} + } + c.mu.Lock() + c.wants = held + c.mu.Unlock() + a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, Kept: kept, Placements: given}, fps, running, head, rev) + return a, true +} + +func sameAnswer(a, b delivery.CandidateAnswer) bool { + x, errA := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": a}}) + y, errB := delivery.EncodeHaveNode(delivery.HaveNode{Candidates: map[string]delivery.CandidateAnswer{"g": b}}) + return errA == nil && errB == nil && string(x) == string(y) +}