diff --git a/cmd/mesh-host/engineseat.go b/cmd/mesh-host/engineseat.go index 772ff55..443f0fc 100644 --- a/cmd/mesh-host/engineseat.go +++ b/cmd/mesh-host/engineseat.go @@ -365,20 +365,24 @@ func packagesFrom(node string, session *link.Session, composer *atomic.Pointer[s // composeCandidateFrom answers the verb compose-candidate (novox/hq ADR 0593 §7): the check's candidate composed on this // node in shadow by the run's shadow, applying and writing nothing, within delivery.ComposeAnswerBound. The answer is // the seat's shape (delivery.ComposeCandidateAnswer), checked as it is encoded; a node whose shadow does not run yet -// refuses in words, which the checker takes as no answer. -func composeCandidateFrom(shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) { +// answers unready, which the checker takes as no answer. +func composeCandidateFrom(node string, shadow *atomic.Pointer[selfcompose.Shadow]) func(candidate string, input bool) (any, error) { return func(candidate string, input bool) (any, error) { ask := delivery.ComposeCandidateAsk{Candidate: candidate, Input: input} if _, err := delivery.EncodeComposeCandidateAsk(ask); err != nil { return nil, err } - s := shadow.Load() - if s == nil { - return nil, errors.New("this node's shadow does not run now (its node agent is starting), so it composes no candidate") + var a delivery.ComposeCandidateAnswer + if s := shadow.Load(); s == nil { + // Unready, never an error the checker reads words of: its agent is starting (ADR 0593 §9). + a = delivery.ComposeCandidateAnswer{Node: node, Candidate: candidate, Verdict: delivery.ComposeCandidateUnready, + Refused: "this node's shadow does not run now (its node agent is starting), so it composes no candidate"} + } else { + ctx, cancel := context.WithTimeout(context.Background(), delivery.ComposeAnswerBound) + defer cancel() + a = s.ComposeCandidate(ctx, ask) } - ctx, cancel := context.WithTimeout(context.Background(), delivery.ComposeAnswerBound) - defer cancel() - raw, err := delivery.EncodeComposeCandidateAnswer(s.ComposeCandidate(ctx, ask)) + raw, err := delivery.EncodeComposeCandidateAnswer(a) if err != nil { return nil, fmt.Errorf("this node's answer to %s does not fit the seat's shape: %w", candidate, err) } diff --git a/cmd/mesh-host/main.go b/cmd/mesh-host/main.go index 1bd7840..b352d48 100644 --- a/cmd/mesh-host/main.go +++ b/cmd/mesh-host/main.go @@ -1001,7 +1001,7 @@ func runLink(ctx context.Context, opts options) error { queue.Acts.Composition = compositionFrom(current) queue.Acts.Compose = composeFrom(current, &composerNow) queue.Acts.Packages = packagesFrom(mine.Node, queue, &composerNow) - queue.Acts.ComposeCandidate = composeCandidateFrom(&shadowNow) + queue.Acts.ComposeCandidate = composeCandidateFrom(mine.Node, &shadowNow) setRunsHere(runsFromAgent(current)) factsOfNow := func(ctx context.Context, ag *agent.Agent) { sayFacts(ctx, opts, mine.Membership.Signer, queue, say, ag) diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 17eab9b..e5518af 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -507,10 +507,22 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str h := b.have("web") return h.FP == fp && h.Composed != "" && h.Phase == delivery.PhaseRunning }) - if keys, _ := b.composed.Keys(); len(keys) != 0 { - t.Errorf("the controller's compositions hold %v; the node composed its own", keys) - } + // ADR 0593 §6: the node publishes what it composed in its composition state, under its own grant: web at the build + // its want names, with the digest its have names (the plan left out of it), and the node's own. first := b.have("web").Composed + webKey, _ := delivery.ComposedKey(itNode, "web", fp) + within(t, 5*time.Second, "the node did not publish web's composition in its composition state", func() bool { + e, err := b.composed.Get(webKey) + if err != nil { + return false + } + c, err := delivery.DecodeComposed(e.Value()) + if err != nil { + return false + } + d, err := delivery.ComposedDigestOf(c) + return err == nil && d == first && c.Signature == nil + }) if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"salut","source":"module"}`)); err != nil { t.Fatal(err) } @@ -551,4 +563,78 @@ func h8rest(t *testing.T, b *itBed, resolved, candidates nats.KeyValue, hkey str 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")) } + h8candidate(t, b, fp, second) +} + +// h8candidate is ADR 0593 §7: a check's candidate is composed on the node through its seat's verb compose-candidate, +// under its own grant: the indexer's candidate changing web's manifest is answered composed, web changed at the +// candidate's build, within the answer bound; nothing is applied and the composition state does not move. +func h8candidate(t *testing.T, b *itBed, fp, running string) { + t.Helper() + kvOf := func(bucket string) nats.KeyValue { + kv, err := b.js.CreateKeyValue(&nats.KeyValueConfig{Bucket: bucket, History: 5}) + if err != nil { + t.Fatal(err) + } + return kv + } + cands, _ := kvOf(delivery.CandidatesBucket), kvOf(delivery.VerdictsBucket) + manifests, err := b.js.KeyValue(delivery.ManifestsBucket) + if err != nil { + t.Fatal(err) + } + path := filepath.Join(b.dir, "web.conf") + man := `{"module":"web","version":"2","config":{"greeting":{"type":"string","description":"what the file says","default":"hello"}},` + + `"resources":[{"id":"conf","type":"file","path":"` + path + `","mode":"0640","content":"greeting=${config:greeting}\n"},` + + `{"id":"tools","type":"package","package":"web-tools"}]}` + digest := delivery.ManifestDigest([]byte(man)) + mraw, _ := delivery.EncodeManifest(delivery.Manifest{Module: "web", JSON: man}) + mkey, _ := delivery.ManifestsKey(digest) + if _, err := manifests.Put(mkey, mraw); err != nil { + t.Fatal(err) + } + head := delivery.Head{Repository: "novox/web", Number: 3, Head: strings.Repeat("3", 40)} + heads, _ := delivery.HeadsHash([]delivery.Head{head}) + pulls := []delivery.CandidatePull{{Head: head, Modules: []string{"web"}, + Fps: map[string]delivery.CandidateModule{"web": {FP2: fpIt("c"), Manifest: digest}}}} + craw, err := delivery.EncodeCandidate(delivery.Candidate{Group: "novox/web#3", HeadsHash: heads, Pulls: pulls, + FPSet: delivery.FingerprintSet(pulls)}) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.CompositionKey("novox/web#3", heads) + if _, err := cands.Put(key, craw); err != nil { + t.Fatal(err) + } + before, _ := b.composed.Status() + got, err := composeCandidateFrom(itNode, &shadowNow)(key, false) + if err != nil { + t.Fatal(err) + } + raw, _ := got.(json.RawMessage) + a, err := delivery.DecodeComposeCandidateAnswer(raw) + if err != nil { + t.Fatalf("the answer %s: %v", raw, err) + } + if a.Verdict != delivery.ComposeCandidateComposed || len(a.Modules) != 1 || a.Modules[0].Module != "web" || + a.Modules[0].FP != fpIt("c") || !a.Modules[0].Changed || a.Took > delivery.ComposeAnswerBound.Milliseconds() { + t.Errorf("the node's answer to a candidate changing web: %+v", a) + } + after, _ := b.composed.Status() + if h := b.have("web"); h.Composed != running || b.file("web") != "greeting=salut\n" || before == nil || after == nil || + after.Values() != before.Values() { + t.Errorf("composing a candidate moved what runs or the composition state: %+v, %q", h, b.file("web")) + } + if _, err := b.composed.Get(mustComposedKey(t, "web", fpIt("c"))); err == nil { + t.Error("the candidate's composition was written into the composition state") + } +} + +func mustComposedKey(t *testing.T, module, fp string) string { + t.Helper() + key, err := delivery.ComposedKey(itNode, module, fp) + if err != nil { + t.Fatal(err) + } + return key } diff --git a/internal/nodegrant/testdata/node.json b/internal/nodegrant/testdata/node.json index 92e72e8..16ce048 100644 --- a/internal/nodegrant/testdata/node.json +++ b/internal/nodegrant/testdata/node.json @@ -7,8 +7,10 @@ "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_placement.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-assigner_wants.*.$KV.mesh-delivery-assigner_wants.want.NODE.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-builder_builds.>", + "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_candidates.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_manifests.>", "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_modules.>", + "$JS.API.CONSUMER.CREATE.KV_mesh-delivery-indexer_verdicts.>", "$JS.API.CONSUMER.CREATE.KV_mesh-host_acts.*.$KV.mesh-host_acts.NODE.>", "$JS.API.CONSUMER.CREATE.KV_mesh-host_data.*.$KV.mesh-host_data.NODE.>", "$JS.API.CONSUMER.CREATE.KV_mesh-host_health.*.$KV.mesh-host_health.NODE", @@ -23,8 +25,10 @@ "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_placement.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_wants.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-builder_builds.>", + "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-indexer_candidates.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-indexer_manifests.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-indexer_modules.>", + "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-indexer_verdicts.>", "$JS.API.CONSUMER.DELETE.KV_mesh-host_acts.>", "$JS.API.CONSUMER.DELETE.KV_mesh-host_data.>", "$JS.API.CONSUMER.DELETE.KV_mesh-host_health.>", @@ -73,8 +77,10 @@ "$JS.FC.KV_mesh-delivery-assigner_placement.>", "$JS.FC.KV_mesh-delivery-assigner_wants.>", "$JS.FC.KV_mesh-delivery-builder_builds.>", + "$JS.FC.KV_mesh-delivery-indexer_candidates.>", "$JS.FC.KV_mesh-delivery-indexer_manifests.>", "$JS.FC.KV_mesh-delivery-indexer_modules.>", + "$JS.FC.KV_mesh-delivery-indexer_verdicts.>", "$JS.FC.KV_mesh-host_acts.>", "$JS.FC.KV_mesh-host_data.>", "$JS.FC.KV_mesh-host_health.>", diff --git a/internal/selfcompose/candidate.go b/internal/selfcompose/candidate.go index c215cc0..de8eca4 100644 --- a/internal/selfcompose/candidate.go +++ b/internal/selfcompose/candidate.go @@ -40,18 +40,25 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi a.Took = time.Since(began).Milliseconds() return a } + // A node that cannot compose now (its shadow not ready, a read the bus did not answer, the bound passed) answers + // unready: the checker takes it as no answer (ADR 0593 §9) by the verdict, the words only saying why. + unready := func(why string) delivery.ComposeCandidateAnswer { + a = refuse(why) + a.Verdict = delivery.ComposeCandidateUnready + return a + } if _, _, err := delivery.ParseCompositionKey(ask.Candidate); err != nil { return refuse(err.Error()) } ctx, cancel := context.WithTimeout(ctx, delivery.ComposeAnswerBound) defer cancel() if !s.ready.Load() { - return refuse(errNotReady.Error()) + return unready(errNotReady.Error()) } base := c.scratch() raw, err := base.get(ctx, delivery.CandidatesBucket, ask.Candidate) if err != nil { - return refuse(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err)) + return unready(fmt.Sprintf("the indexer's candidate %s cannot be read: %v", ask.Candidate, err)) } if raw == nil { return refuse("the indexer holds no candidate " + ask.Candidate) @@ -63,7 +70,7 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi laid := c.scratch() touched, err := laid.layCandidate(ctx, cand) if err != nil { - return refuse(err.Error()) + return unready(err.Error()) } var running []delivery.Composed @@ -89,11 +96,11 @@ func (s *Shadow) ComposeCandidate(ctx context.Context, ask delivery.ComposeCandi } }) if ctx.Err() != nil { - return refuse(fmt.Sprintf("the bound of %s passed before the node composed the candidate", delivery.ComposeAnswerBound)) + return unready(fmt.Sprintf("the bound of %s passed before the node composed the candidate", delivery.ComposeAnswerBound)) } switch { case laidErr != nil && BusFailed(laidErr): - return refuse("a read on the bus did not answer: " + laidErr.Error()) + return unready("a read on the bus did not answer: " + laidErr.Error()) case laidErr != nil && baseErr != nil: // The node composes none of its own without the candidate either: the candidate changes nothing of that. a.Verdict, a.Modules = delivery.ComposeCandidateComposed, []delivery.ModuleComposed{} diff --git a/internal/selfcompose/candidate_test.go b/internal/selfcompose/candidate_test.go index a01ab32..aec1323 100644 --- a/internal/selfcompose/candidate_test.go +++ b/internal/selfcompose/candidate_test.go @@ -16,7 +16,7 @@ import ( // changes a placed module composes it at the candidate's build, changed; one whose manifest the indexer refused is // answered refused, naming the module and the judge's words, and the verdict is refused; a candidate the indexer does not // hold is refused in words. Nothing is applied, nothing is written, no placement answer is handed, and the answer comes -// within the bound (an ended bound answers at once, refused). +// within the bound; a shadow not ready and an ended bound answer unready at once, which the checker takes as no answer. func TestTheNodeComposesACandidateOnAsk(t *testing.T) { b := newFakeBus() fp := delivery.SourcePrefix2 + strings.Repeat("a", 64) @@ -71,7 +71,7 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() go func() { _ = s.Run(ctx) }() - if a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted}); !strings.Contains(a.Refused, "not read every input") { + if a := s.ComposeCandidate(ctx, delivery.ComposeCandidateAsk{Candidate: accepted}); a.Verdict != delivery.ComposeCandidateUnready || a.Refused == "" { t.Errorf("a shadow not ready answered %+v", a) } place, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Decided: delivery.NodeDecided{Hub: true, Address: "10.77.0.1"}, @@ -116,7 +116,7 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) { } ended, end := context.WithCancel(ctx) end() - if e := s.ComposeCandidate(ended, delivery.ComposeCandidateAsk{Candidate: accepted}); e.Verdict != delivery.ComposeCandidateRefused { + if e := s.ComposeCandidate(ended, delivery.ComposeCandidateAsk{Candidate: accepted}); e.Verdict != delivery.ComposeCandidateUnready { t.Errorf("an ended bound answered %+v", e) } if len(state.written()) != 0 || applied.Load() != 0 || handed.Load() != 0 {