From 23799a30ed4df1960057738adc7fc913a0cc3431 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 23:05:06 +0100 Subject: [PATCH] R3-1, R3-2, R3-3: the node-engine reads no build of the builder; the shadow builds from the entry it holds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The composer's Build is made from the catalog half's entry it holds and the parts of the want that names it (buildOf); the inputs of the indexer's modules and the builder's builds, takeBuildDone, takeModule and the builder read are gone, and mesh-host names no BuilderBuildsBucket. The shadow takes the entries of the builds it composes from (askBuildsEntries), builds a module a candidate adds from its entry and the candidate's `added` parts (A-316), and holds, with the reason, a module whose entry has no part given, whose fingerprint the verb module does not give, or for which it errors (A-312). Its composer is given the entry call as the node's own is. Rows: a built module added with and without parts, the verb module answering no fingerprint and erroring. Builds: design 47 G §421, §628, A-308, A-312, A-316 --- cmd/mesh-host/nodeagent.go | 6 +- cmd/mesh-host/nodeagent_integration_test.go | 15 +- internal/compose/frombus/frombus.go | 2 +- internal/selfcompose/candidate.go | 4 +- .../selfcompose/compose_while_passing_test.go | 3 +- internal/selfcompose/composedstate_test.go | 90 +++----- internal/selfcompose/data_test.go | 38 ---- internal/selfcompose/entries.go | 42 +++- internal/selfcompose/entries_test.go | 69 +++++- internal/selfcompose/selfcompose.go | 128 ++---------- internal/selfcompose/selfcompose_test.go | 197 +++++++----------- internal/selfcompose/shadow.go | 44 +++- 12 files changed, 279 insertions(+), 359 deletions(-) diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 75310ea0..8a256529 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -1155,9 +1155,6 @@ func selfComposer(session *link.Session, node string, hooks composerHooks, say f env.ComposedState = link.EngineComposed{Session: session, Node: node} // And each composition it writes there said as the seat's event composed, which the assigner waits on. env.SayComposed = session.SayComposed - // And the catalog half read by the fingerprints its wants and runnings name, over its own calls of the seat mesh-catalog's - // verb entry on its link session (design 47 G §421): neither the indexer's nor the builder's states are read. - env.Entry = selfcompose.EntryCall(session.AskMeshSeat) return selfcompose.New(env) } @@ -1199,6 +1196,9 @@ func composerEnv(session *link.Session, node string, say func(string), hooks com } else if hooks.seatListFailed { env.SeatList = selfcompose.Unread{} } + // And the shadow too: the catalog half read by the fingerprints its wants and runnings name, over its own calls of the seat mesh-catalog's + // verb entry on its link session (design 47 G §421): neither the indexer's nor the builder's states are read. + env.Entry = selfcompose.EntryCall(session.AskMeshSeat) return env } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index 05b9eed8..933bbaac 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -12,7 +12,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" condseat "git.novox.be/novox/mesh-contracts/health-and-repair/go/seats/meshconditions" - builderbuilds "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshdeliveryclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" @@ -493,7 +492,6 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { } assignment, modules, manifests := kvOf(buckets.Assignments), kvOf(buckets.Modules), kvOf(buckets.Manifests) resolved := kvOf(selfcompose.ResolvedBucket) - builds := kvOf(builderbuilds.BuilderBuildsBucket) candidates := kvOf(buckets.ProposalCandidates) packages := kvOf(link.PackagesBucket) path := filepath.Join(b.dir, "web.conf") @@ -523,10 +521,6 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { rec2, _ := nec.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "it2", Capabilities: []string{"service-manager"}}}) pkey2, _ := meshmoduleassignerclient.NodeKey("it2") hkey2, _ := nec.HaveNodeKey("it2") - buildKey, _ := builderbuilds.BuilderBuildKey("web", fp) - done, _ := builderbuilds.EncodeBuilderBuild(builderbuilds.BuilderBuild{State: builderbuilds.BuildDone, Manifest: digest, - Artifacts: []builderbuilds.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://web/files/blobs/" + fpDigest("f"), Digest: fpDigest("f")}}}) - firstBuild := done for _, w := range []struct { kv nats.KeyValue key string @@ -534,7 +528,7 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { }{{manifests, mkey, mraw}, {modules, "web", modv}, {b.haves, hkey, rec}, {resolved, itNode + ".web.greeting", []byte(`{"value":"bonjour","source":"module"}`)}, {manifests, mlkey, mlraw}, {modules, "mailer", mlmod}, {b.haves, hkey2, rec2}, {assignment, pkey2, places2[pkey2]}, {assignment, "it2.mailer", places2["it2.mailer"]}, - {resolved, "it2.mailer.domain", []byte(`{"value":"example.org","source":"module"}`)}, {builds, buildKey, firstBuild}} { + {resolved, "it2.mailer.domain", []byte(`{"value":"example.org","source":"module"}`)}} { if _, err := w.kv.Put(w.key, w.value); err != nil { t.Fatal(err) } @@ -566,7 +560,6 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { t.Fatal(err) } if checkBuild { - _ = buildKey // The entry of the build its want names is asked of the catalog half through the node's session: one the half holds none // of holds the want (entry-missing, naming the fingerprint) and nothing is applied; once the half holds it and it is // asked again by hand, then a config event, and the node composes. @@ -855,7 +848,6 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test } assignment, modules, manifests := kvOf(buckets.Assignments), kvOf(buckets.Modules), kvOf(buckets.Manifests) kvOf(selfcompose.ResolvedBucket) - builds := kvOf(builderbuilds.BuilderBuildsBucket) kvOf(buckets.ProposalCandidates) kvOf(link.PackagesBucket) // The bus module at a build, done: its manifest written and the indexer's module naming it at fp. @@ -872,11 +864,6 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test if _, err := manifests.Put(mkey, mraw); err != nil { t.Fatal(err) } - bkey, _ := builderbuilds.BuilderBuildKey("nats", fp) - done, _ := builderbuilds.EncodeBuilderBuild(builderbuilds.BuilderBuild{State: builderbuilds.BuildDone, Manifest: digest}) - if _, err := builds.Put(bkey, done); err != nil { - t.Fatal(err) - } if _, err := modules.Put("nats", modv); err != nil { t.Fatal(err) } diff --git a/internal/compose/frombus/frombus.go b/internal/compose/frombus/frombus.go index 4f072247..5c85ada0 100644 --- a/internal/compose/frombus/frombus.go +++ b/internal/compose/frombus/frombus.go @@ -81,7 +81,7 @@ type Reader interface { // and each it grants as a provider, with the copy sealed to it (the vault's `own` with the keys, ADR 0550 §5). A key // the vault holds no value for is left out; the composition then stands in for it, and Compose refuses it. Credentials(ctx context.Context, node string, asConsumer, asProvider []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) - // Build is a module's build at a fingerprint, the builder's key (builds.BuilderBuildsBucket, BuilderBuildKey); nil + // Build is a module's build at a fingerprint, the builder's key (BuilderBuildKey); nil // when the builder holds none. The composing node composes each of its modules from its build (issue 1126). Build(ctx context.Context, module, fp string) (*builds.BuilderBuild, error) // Archive is the archive at url, a kept reference routed to the artifact store as the composing node reaches it diff --git a/internal/selfcompose/candidate.go b/internal/selfcompose/candidate.go index 53161098..643bed97 100644 --- a/internal/selfcompose/candidate.go +++ b/internal/selfcompose/candidate.go @@ -322,7 +322,9 @@ func (c *Composer) scratch() *Composer { out.values = maps.Clone(c.values) out.manifests = maps.Clone(c.manifests) out.archives = maps.Clone(c.archives) - out.builds = maps.Clone(c.builds) + // The entries it holds and the wants that name them give a build (buildOf), as the builder's builds once did. + out.keptEnt = maps.Clone(c.keptEnt) + out.wants = maps.Clone(c.wants) out.sealedOwn = maps.Clone(c.sealedOwn) out.ownHanded = maps.Clone(c.ownHanded) out.certified = c.certified diff --git a/internal/selfcompose/compose_while_passing_test.go b/internal/selfcompose/compose_while_passing_test.go index fe2681f2..d755ae2d 100644 --- a/internal/selfcompose/compose_while_passing_test.go +++ b/internal/selfcompose/compose_while_passing_test.go @@ -10,7 +10,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "github.com/novox/mesh-host/internal/agent" "github.com/novox/mesh-host/internal/buckets" @@ -48,7 +47,7 @@ func composeAnsweredWhileAPassRuns(t *testing.T) { b.putAssigned(t, place) b.put(t, buckets.Have, "n1.node", have) b.put(t, buckets.Wants, "want.n1.node", nodeWant) - for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } go func() { diff --git a/internal/selfcompose/composedstate_test.go b/internal/selfcompose/composedstate_test.go index 68f9721d..5de5c96f 100644 --- a/internal/selfcompose/composedstate_test.go +++ b/internal/selfcompose/composedstate_test.go @@ -63,13 +63,31 @@ func (f *fakeComposed) Delete(_ context.Context, key string) error { func (f *fakeComposed) written() []string { f.mu.Lock() defer f.mu.Unlock() - return slices.Clone(f.puts) + var out []string + for _, key := range f.puts { + // The entries and wants the node keeps are its composition state's too (§683), not compositions. + if !strings.HasPrefix(key, keptEntryPrefix) && !strings.HasPrefix(key, keptWantPrefix) { + out = append(out, key) + } + } + return out +} + +// keep holds in the state, as an earlier run of the node left them, the entry of fp and the want of module at it (§683): the +// composer then composes at its start from what it kept, and asks the catalog half of nothing first. +func (f *fakeComposed) keep(t *testing.T, half catalogHalf, module, fp string) { + t.Helper() + entry, _ := json.Marshal(meshcatalog.EntryAnswer{Entry: half[fp]}) + want, _ := json.Marshal(keptWant{V: 1, Since: time.Now(), Want: meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true}}) + f.mu.Lock() + defer f.mu.Unlock() + f.kv[entryKey(fp)] = ComposedEntry{Value: entry, Revision: 1} + f.kv[wantStateKey(module)] = ComposedEntry{Value: want, Revision: 1} } // novox/hq ADR 0593 §6: the node publishes each composition it makes in its composition state, a record never applied: -// the wanted module and the node's own at the first pass; nothing again for a pass that changes nothing; a done build -// of a module placed here that no want names, at its trunk's build, written once as its own key (before any want moves); -// the same build landing again writes nothing. And a module's keys are kept while among the five it wrote last or named +// the wanted module and the node's own at the first pass, from the entry and want it kept; nothing again for a pass that +// changes nothing; a failed build of a placed module said refused by its build-ended event. And a module's keys are kept while among the five it wrote last or named // by a want, and deleted otherwise. func TestTheNodePublishesItsCompositions(t *testing.T) { t.Run("the verb compose is answered while a pass runs", composeAnsweredWhileAPassRuns) @@ -90,6 +108,12 @@ func TestTheNodePublishesItsCompositions(t *testing.T) { state := &fakeComposed{kv: map[string]ComposedEntry{}} env := b.env() env.ComposedState = state + ofp0 := meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64) + half := catalogHalf{} + half.add(t, "lemurs", fp, lemurs) + half.add(t, "oldie", ofp0, oldie) + env.Entry = half.call() + state.keep(t, half, "lemurs", fp) // Design 47, *What each step says*: each module composition written says composed, with its fingerprint and digest. var saidMu sync.Mutex var saidComposed []string @@ -124,13 +148,12 @@ func TestTheNodePublishesItsCompositions(t *testing.T) { } b.put(t, buckets.Wants, "want.n1.lemurs", want) b.put(t, buckets.Wants, "want.n1.node", nodeWant) - for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } for u := next(t, out); !u.Caught; u = next(t, out) { } lemursKey, _ := engcomp.Key("n1", "lemurs", fp) - oldieKey, _ := engcomp.Key("n1", "oldie", ofp) waitFor := func(what string, ok func([]string) bool) []string { t.Helper() deadline := time.Now().Add(5 * time.Second) @@ -190,60 +213,6 @@ func TestTheNodePublishesItsCompositions(t *testing.T) { if got := state.written(); len(got) != 2 { t.Errorf("a pass that changed nothing wrote %v", got[2:]) } - // oldie's trunk build is done: its key at that build is written once, though no want names it. - build, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: odigest}) - bkey, _ := builds.BuilderBuildKey("oldie", ofp) - b.mu.Lock() - b.kv[builds.BuilderBuildsBucket+"/"+bkey] = build - b.mu.Unlock() - b.put(t, builds.BuilderBuildsBucket, bkey, build) - got := waitFor("oldie's done build", func(got []string) bool { return len(got) >= 3 }) - if got[2] != oldieKey { - t.Errorf("the done build of oldie wrote %v, want %s", got[2:], oldieKey) - } - b.put(t, builds.BuilderBuildsBucket, bkey, build) - time.Sleep(100 * time.Millisecond) - if got := state.written(); len(got) != 3 { - t.Errorf("the same build landing again wrote %v", got[3:]) - } - - // A placed module whose newest build failed is published as a refusal at that build with the build's words (the - // provisioner holds it and does not wait); one whose build is queued or running stays unpublished. - for m, st := range map[string]builds.BuilderBuild{ - "broken": {State: builds.BuildFailed, Error: `refused kind: package of language "go"`}, - "waiting": {State: builds.BuildQueued}, - } { - f := map[string]string{"broken": bfp, "waiting": wfp}[m] - raw, _ := builds.EncodeBuilderBuild(st) - k, _ := builds.BuilderBuildKey(m, f) - b.mu.Lock() - b.kv[builds.BuilderBuildsBucket+"/"+k] = raw - b.mu.Unlock() - b.put(t, builds.BuilderBuildsBucket, k, raw) - } - brokenKey, _ := engcomp.Key("n1", "broken", bfp) - got = waitFor("broken's failed build", func(got []string) bool { return len(got) >= 4 }) - // The pass that wrote it says so ("1 key(s) written" was said once before, for oldie); then only brokenKey is new. - for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { - n := 0 - b.mu.Lock() - for _, l := range b.said { - if strings.Contains(l, "this node's composition state: 1 key(s) written") { - n++ - } - } - b.mu.Unlock() - if n >= 2 { - break - } - } - if got := state.written(); len(got) != 4 || got[3] != brokenKey { - t.Errorf("a failed build and a queued one wrote %v, want only %s", got[3:], brokenKey) - } - if comp, err := engcomp.Decode(state.kv[brokenKey].Value); err != nil || comp.Refused != `refused kind: package of language "go"` || len(comp.Resources) != 0 { - t.Errorf("the failed build's key: %+v, %v", comp, err) - } - // What is kept: the five keys of a module written last, and one a want names; the rest is deleted. keep := New(b.env()) keep.pubKept = map[string]ComposedEntry{} @@ -262,6 +231,7 @@ func TestTheNodePublishesItsCompositions(t *testing.T) { t.Errorf("past keeping with a want naming the oldest: %v, want none", past) } + brokenKey, _ := engcomp.Key("n1", "broken", bfp) // Design 47 G A-293: reading the catalog half by fingerprint, a placed module's newest build that ended is the last trunk // build.ended heard of it; a failed one is said refused with the event's words once the first read ended, an end naming a // branch moves nothing, a module with no end (queued or running) stays unpublished, and a later done end, at any diff --git a/internal/selfcompose/data_test.go b/internal/selfcompose/data_test.go index 2f61b889..41c90dfe 100644 --- a/internal/selfcompose/data_test.go +++ b/internal/selfcompose/data_test.go @@ -5,7 +5,6 @@ import ( "errors" "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "github.com/novox/mesh-host/internal/buckets" engcomp "github.com/novox/mesh-host/internal/compositions" @@ -17,8 +16,6 @@ import ( "git.novox.be/novox/mesh-sdk/go/catalogue" "github.com/novox/mesh-host/internal/compose/frombus" - - "github.com/novox/mesh-host/internal/agent" ) // fakeData is the engine's state data as a map, counting the writes and deletes made. @@ -231,39 +228,4 @@ func TestTheEngineKeepsEachPlacedModulesDataItems(t *testing.T) { if wrote, _ := pc.WritePackages(context.Background()); wrote { t.Error("the verb's write of a key holding what was composed wrote it again") } - - // The items are worked out from what the pass read: a done build is read over the bus once, not again for the data - // items after the pass (a module in a running build is read again, since it may land). - b := newFakeBus() - done, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: meshcatalogclient.ManifestDigest([]byte(lemurs))}) - running, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: "building"}) - fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64) - doneKey, _ := builds.BuilderBuildKey("web", fp) - runningKey, _ := builds.BuilderBuildKey("pg", fp) - b.kv[builds.BuilderBuildsBucket+"/"+doneKey], b.kv[builds.BuilderBuildsBucket+"/"+runningKey] = done, running - r := reader{c: New(b.env())} - for range 3 { - if got, err := r.Build(context.Background(), "web", fp); err != nil || got.State != builds.BuildDone { - t.Fatalf("the done build: %+v, %v", got, err) - } - if got, err := r.Build(context.Background(), "pg", fp); err != nil || got.State != "building" { - t.Fatalf("the running build: %+v, %v", got, err) - } - } - reads := map[string]int{} - for _, read := range b.reads { - reads[read[1]]++ - } - if reads[doneKey] != 1 || reads[runningKey] != 3 { - t.Errorf("the builds were read %v times over the bus; want the done one once, the running one each time", reads) - } - // A done key the builder claims again (a gone step, issue 1937) and lands done again is read again, not served kept. - again, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: meshcatalogclient.ManifestDigest([]byte("{}"))}) - b.mu.Lock() - b.kv[builds.BuilderBuildsBucket+"/"+doneKey] = again - b.mu.Unlock() - r.c.takeBuildDone(agent.Update{Key: doneKey, Value: running}) - if got, err := r.Build(context.Background(), "web", fp); err != nil || got.Manifest != meshcatalogclient.ManifestDigest([]byte("{}")) { - t.Errorf("a done build written again is served as %+v (%v), not read again", got, err) - } } diff --git a/internal/selfcompose/entries.go b/internal/selfcompose/entries.go index 8c427341..c45daa2e 100644 --- a/internal/selfcompose/entries.go +++ b/internal/selfcompose/entries.go @@ -7,7 +7,9 @@ import ( "encoding/json" "fmt" "git.novox.be/novox/mesh-sdk/go/buscalls" + "maps" "reflect" + "slices" "sort" "strings" "time" @@ -441,9 +443,10 @@ func referenceOf(module, artifact, digest string) string { return "artifact-store://" + module + "/" + artifact + "/blobs/" + digest } -// buildOf is the build a want names, made from its entry and its parts alone (the builder's done key is not read): done, with -// each artifact of the entry at the digest its part names and the reference made from it. Nil where the entry is not held, the -// want names another fingerprint, or a part is missing. +// buildOf is the build a want names, made from the entry the composer holds and the parts of the want (the builder's done key is +// not read, §421, §628): done, with each artifact of the entry at the digest its part names and the reference made from it. Nil +// where the entry is not held, the want names another fingerprint, or a part is missing. The shadow's wants for the modules a +// candidate adds carry the parts the candidate gives (A-316). func (c *Composer) buildOf(module, fp string) *builds.BuilderBuild { c.mu.Lock() w, wanted := c.wants[module] @@ -640,3 +643,36 @@ func (c *Composer) addedModule(ctx context.Context, module string) (fp string, e } return fp, e, false, true, fmt.Sprintf("the entry of %s's build %s is asked of the catalog half and not yet answered", module, fp) } + +// askBuildsEntries asks the entry of each build in fps, the builds the node runs or its wants name, that the composer does not hold, +// and says the first not yet answered: the shadow composes a candidate from entries alone (§421, §628), and an answer wakes it. The +// fingerprints are kept beside the candidates' added ones while a proposal stands (addedFps). +func (c *Composer) askBuildsEntries(fps map[string]string) (pending string) { + if !c.byEntries() { + return "" + } + for _, module := range slices.Sorted(maps.Keys(fps)) { + fp := fps[module] + if !strings.HasPrefix(fp, meshcatalogclient.SourcePrefix2) { + continue + } + c.mu.Lock() + if c.addedFps == nil { + c.addedFps = map[string]bool{} + } + c.addedFps[fp] = true + c.mu.Unlock() + e, state := c.entryOf(fp) + switch { + case state == meshcatalogclient.EntryAnswered: + c.mu.Lock() + if _, kept := c.keptEnt[fp]; !kept && e.Module != "" { + c.keptEnt[fp] = keptEntry{entry: e, since: c.now()} + } + c.mu.Unlock() + case state == meshcatalogclient.EntryAsked && pending == "": + pending = fmt.Sprintf("the entry of %s's build %s is asked of the catalog half and not yet answered", module, fp) + } + } + return pending +} diff --git a/internal/selfcompose/entries_test.go b/internal/selfcompose/entries_test.go index 15b6dd4e..b7fb2ac1 100644 --- a/internal/selfcompose/entries_test.go +++ b/internal/selfcompose/entries_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "strings" "sync" "testing" @@ -12,7 +13,6 @@ import ( "git.novox.be/novox/mesh-contracts/go/clients/nodeengineclient" "git.novox.be/novox/mesh-contracts/go/seats/meshmoduleassigner" nodeengine "git.novox.be/novox/mesh-contracts/go/seats/nodeengine" - "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/clients/meshcatalogclient" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshcatalog" "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" @@ -97,7 +97,7 @@ func composeFromEntries(t *testing.T) { // Neither the indexer's states nor the builder's are watched or read. b.mu.Lock() for _, r := range b.reads { - if r[0] == buckets.Modules || r[0] == buckets.Manifests || r[0] == buckets.Verdicts || r[0] == builds.BuilderBuildsBucket { + if r[0] == buckets.Modules || r[0] == buckets.Manifests || r[0] == buckets.Verdicts || strings.HasSuffix(r[0], "_builds") { t.Errorf("the node-engine read %s of %s", r[1], r[0]) } } @@ -183,3 +183,68 @@ func composeFromEntries(t *testing.T) { t.Errorf("a removed want left its kept want %v and entry %v", hasWant, hasEntry) } } + +// takeModule lays one module of the catalog half's picture as the composer's own lay of an entry does (layModules): the tests +// that compose from a picture they set read no indexer. +func (c *Composer) takeModule(u agent.Update) { + name, err := meshcatalogclient.ParseModulesKey(u.Key) + if err != nil { + return + } + c.mu.Lock() + defer c.mu.Unlock() + m, err := meshcatalogclient.DecodeModule(u.Value) + if u.Deleted || err != nil { + delete(c.modules, name) + return + } + c.modules[name] = m +} + +// catalogHalf is the catalog half's entries a test composes from, by fingerprint, answered through the seat mesh-catalog's verbs +// entry and module (design 47 G §421, A-312): the node-engine reads no indexer. +type catalogHalf map[string]meshcatalog.CatalogEntry + +// add holds the entry of module at fp, made of the manifest body. +func (h catalogHalf) add(t *testing.T, module, fp, body string) { + t.Helper() + var m map[string]any + if err := json.Unmarshal([]byte(body), &m); err != nil { + t.Fatal(err) + } + h[fp] = meshcatalog.CatalogEntry{Module: module, Fp: fp, Manifest: m, Repository: "novox/" + module, Commit: "c", ManifestDigest: meshcatalogclient.ManifestDigest([]byte(body))} +} + +func (h catalogHalf) call() meshcatalogclient.Call { + return EntryCall(func(_ context.Context, _, verb string, args any) (json.RawMessage, error) { + switch in := args.(type) { + case meshcatalog.ModuleInput: + for _, e := range h { + if e.Module == in.Module { + return json.Marshal(meshcatalog.ModuleAnswer{Module: meshcatalog.ModuleAnswerModule{Module: e.Module, Repository: e.Repository, Commit: e.Commit, Fp: e.Fp}}) + } + } + return nil, fmt.Errorf("the catalog holds no module %s", in.Module) + case meshcatalog.EntryInput: + if e, ok := h[in.Fingerprint]; ok { + return json.Marshal(meshcatalog.EntryAnswer{Entry: e}) + } + return nil, errors.New("no entry of fingerprint " + in.Fingerprint + " is kept") + } + return nil, fmt.Errorf("asked %s of %T", verb, args) + }) +} + +// holdEntry makes the composer hold the entry of module at fp, as its call of the verb entry would have answered it (§421), for the +// tests that set the rest of its picture directly. +func (c *Composer) holdEntry(t *testing.T, module, fp, body string) { + t.Helper() + half := catalogHalf{} + half.add(t, module, fp, body) + c.mu.Lock() + defer c.mu.Unlock() + if c.keptEnt == nil { + c.keptEnt = map[string]keptEntry{} + } + c.keptEnt[fp] = keptEntry{entry: half[fp]} +} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index 07999e2c..02b63622 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -180,10 +180,7 @@ type Composer struct { values map[string]map[string]any // by module, this node's manifests map[string][]byte // by digest, read once archives map[string][]byte // templates archives by digest, read once - // builds are the builder's done builds by key, read once: a done build at a fingerprint never changes, so a pass and - // the data items after it (publishData) do not read each placed module's build again over the bus (ADR 0577). - builds map[string]builds.BuilderBuild - handed map[string]string // composed key to the digest handed + handed map[string]string // composed key to the digest handed // notTaken is, by module, why a want of this node's was not taken (it could not be decoded, or names no fingerprint). notTaken map[string]string // unmadeSaid is, by module, the stable part of the reason its last journal line said (without where the picture stands). @@ -203,8 +200,6 @@ type Composer struct { forCandidates bool // addedFps are the builds of modules a candidate adds that the shadow asked the entry of (addedModule), kept beside the wants' under mu. addedFps map[string]bool - // modulesRev is the revision of the last update of the indexer's modules taken: where the picture of them stands. - modulesRev uint64 // heard is, by node and module, what each node's event running last said it runs, kept from the seat's stream by this // node's own consumer (NoteRunning), under mu; heardRead is whether the first read of the stream ended (RunningRead). heard map[string]map[string]heardRun @@ -418,7 +413,7 @@ func New(env Env) *Composer { func newComposer(env Env) *Composer { return &Composer{env: env, wants: map[string]meshwants.WantsByWantNodeModuleValue{}, assignments: map[string]meshmoduleassigner.Assignments{}, assignmentKeys: meshmoduleassignerclient.NewKeys(), records: map[string]nodeengine.NodeRecord{}, inOwn: map[string]bool{}, modules: map[string]meshcatalog.ModulesByModuleValue{}, values: map[string]map[string]any{}, - manifests: map[string][]byte{}, archives: map[string][]byte{}, builds: map[string]builds.BuilderBuild{}, handed: map[string]string{}, notTaken: map[string]string{}, unmadeSaid: map[string]string{}, rehand: make(chan struct{}, 1), sealedOwn: map[string]ownSealed{}, + manifests: map[string][]byte{}, archives: map[string][]byte{}, handed: map[string]string{}, notTaken: map[string]string{}, unmadeSaid: map[string]string{}, rehand: make(chan struct{}, 1), sealedOwn: map[string]ownSealed{}, pkgs: map[string]compositions.Composed{}, asked: make(chan chan Pass, 8)} } @@ -478,22 +473,9 @@ type input struct { changed func(c *Composer, u agent.Update) bool } +// inputs are the buckets the composer watches: it watches neither the indexer's modules nor the builder's builds, but reads the +// catalog half's entry of each build its wants name (§421, §628). func (c *Composer) inputs() []input { - ins := c.allInputs() - if !c.byEntries() { - return ins - } - // Reading the catalog half by fingerprint, it watches neither the indexer's modules nor the builder's builds (§421, §628). - out := ins[:0:0] - for _, in := range ins { - if in.bucket != buckets.Modules && in.bucket != builds.BuilderBuildsBucket { - out = append(out, in) - } - } - return out -} - -func (c *Composer) allInputs() []input { node := c.env.Node return []input{ {buckets.Wants, "want." + node + ".>", (*Composer).takeWant, nil}, @@ -503,50 +485,10 @@ func (c *Composer) allInputs() []input { // A watch of every key is refused by the bus (issue 1126, 2026-10-27 09:21: the fold's `>` stopped every agent). {buckets.Have, "*." + nodeengineclient.NodeWord, nil, (*Composer).takeRecordChanged}, {buckets.Have, node + ".>", nil, (*Composer).takeModuleHave}, - {buckets.Modules, ">", (*Composer).takeModule, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, - // The builder's builds: a pass that held a module for a build not yet done is made again when it lands (issue - // 774, 2026-10-30 19:30: node-tools' build landed and ace, g14 and shanks stayed refused until another input moved). - {builds.BuilderBuildsBucket, ">", nil, (*Composer).takeBuildDone}, } } -// takeBuildDone says whether a build that landed is one this node composes from: done, of a module it wants at that -// build, or wanted or placed here whose trunk's build it is; or failed, of a module placed here whose trunk's build it -// is, which the node publishes as a refusal (ADR 0593 §6). Every other build moves nothing. -func (c *Composer) takeBuildDone(u agent.Update) bool { - // Any write or delete of a build's key forgets the build kept as done: the builder claims a done key again as a new - // revision (a gone step, issue 1937), which lands done with other artifacts. - c.mu.Lock() - delete(c.builds, u.Key) - c.mu.Unlock() - if u.Deleted { - return false - } - module, fp, err := builds.ParseBuilderBuildKey(u.Key) - if err != nil { - return false - } - c.mu.Lock() - w, wanted := c.wants[module] - m := c.modules[module] - placedHere := false - for _, pm := range c.assignments[c.env.Node].Modules { - placedHere = placedHere || pm.Module == module - } - c.mu.Unlock() - // A module placed here is composed at its trunk's build once that is done, for the composition state, before its - // want moves (ADR 0593 §6). - b, err := builds.DecodeBuilderBuild(u.Value) - if err != nil { - return false - } - if b.State == builds.BuildFailed { - return placedHere && m.Fp2 == fp - } - return b.State == builds.BuildDone && (wanted && w.Fp == fp || (wanted || placedHere) && m.Fp2 == fp) -} - // 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. @@ -794,13 +736,6 @@ func (c *Composer) entry(bucket, key string) (v any, has, known bool) { } p, ok := c.assignments[node] return p, ok, true - case buckets.Modules: - name, err := meshcatalogclient.ParseModulesKey(key) - if err != nil { - return nil, false, true - } - m, ok := c.modules[name] - return m, ok, true case ResolvedBucket: rest, ok := strings.CutPrefix(key, c.env.Node+".") if !ok { @@ -969,22 +904,6 @@ func (c *Composer) inOwnOf(p meshmoduleassigner.Assignments) []string { return out } -func (c *Composer) takeModule(u agent.Update) { - name, err := meshcatalogclient.ParseModulesKey(u.Key) - if err != nil { - return - } - c.mu.Lock() - defer c.mu.Unlock() - m, err := meshcatalogclient.DecodeModule(u.Value) - if u.Deleted || err != nil { - delete(c.modules, name) - return - } - c.modules[name] = m - c.modulesRev = u.Revision -} - func (c *Composer) takeValue(u agent.Update) { rest, ok := strings.CutPrefix(u.Key, c.env.Node+".") if !ok { @@ -1659,33 +1578,10 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { return []byte(m.Manifest), nil } -func (r reader) Build(ctx context.Context, module, fp string) (*builds.BuilderBuild, error) { - if r.c.byEntries() && !r.c.forCandidates { - return r.c.buildOf(module, fp), nil - } - key, err := builds.BuilderBuildKey(module, fp) - if err != nil { - return nil, err - } - r.c.mu.Lock() - done, ok := r.c.builds[key] - r.c.mu.Unlock() - if ok { - return &done, nil - } - defer r.c.waiting("the build of " + module + " at " + fp + " (" + builds.BuilderBuildsBucket + ")")() - raw, err := r.c.get(ctx, builds.BuilderBuildsBucket, key) - if err != nil || raw == nil { - return nil, err - } - b, err := builds.DecodeBuilderBuild(raw) - if err != nil { - return nil, fmt.Errorf("the builder's %s: %w", key, err) - } - if b.State == builds.BuildDone { - r.c.keepRead(func(k *Composer) { k.builds[key] = b }) - } - return &b, nil +// Build is the build a want names, made from the catalog half's entry the composer holds (§421, §628): the builder's builds are +// not read. +func (r reader) Build(_ context.Context, module, fp string) (*builds.BuilderBuild, error) { + return r.c.buildOf(module, fp), nil } // archiveWithin bounds one read of a templates archive in the artifact store. @@ -2313,18 +2209,18 @@ func (c *Composer) unmadeOf(fps map[string]string, comps map[string]engcomp.Comp why = fmt.Sprintf("the seat's assignment places no %s on this node (%d modules placed)", module, len(own.Modules)) key = "not placed" case !in: - why = fmt.Sprintf("no manifest in the composer's picture of the indexer's modules (%d modules, last update at revision %d)", - len(c.modules), c.modulesRev) + why = fmt.Sprintf("no manifest in the composer's picture of the indexer's modules (%d modules)", + len(c.modules)) key = "no manifest" case slices.Contains(skipped, module): why = "the composition skipped it: the declaration gives it no fingerprint (indexer's picture: " + m.Fp2 + ")" key = why default: if at := made[module]; at != "" { - why = fmt.Sprintf("the pass composed it at %s and the want names %s (indexer's picture: %s, revision %d)", at, fp, m.Fp2, c.modulesRev) + why = fmt.Sprintf("the pass composed it at %s and the want names %s (indexer's picture: %s)", at, fp, m.Fp2) key = "composed at " + at } else { - why = fmt.Sprintf("the composition made nothing of it at %s (indexer's picture: %s, revision %d)", fp, m.Fp2, c.modulesRev) + why = fmt.Sprintf("the composition made nothing of it at %s (indexer's picture: %s)", fp, m.Fp2) key = "made nothing at " + m.Fp2 } } diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index 4face352..0620df94 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -210,11 +210,14 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { b.put(t, buckets.Have, "n1.node", have) oldHave, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) b.put(t, buckets.Have, "n1.oldie", oldHave) - b.put(t, buckets.Modules, "lemurs", mod) - b.put(t, buckets.Modules, "oldie", omod) + // The picture of the catalog half is set directly (no indexer is watched): lemurs's entry, which its want names, and oldie, + // which the node runs inside its own composition and no want names. + c.takeModule(agent.Update{Key: "lemurs", Value: mod}) + c.takeModule(agent.Update{Key: "oldie", Value: omod}) + c.holdEntry(t, "lemurs", fp, lemurs) b.put(t, buckets.Wants, "want.n1.lemurs", want) b.put(t, buckets.Wants, "want.n1.node", nodeWant) - for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } got := map[string]bool{} @@ -636,38 +639,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { b.put(t, ResolvedBucket, "n1.other.k", []byte(`{"value":1,"source":"node"}`)) quiet(t, out) - // Issue 774, 2026-10-30 19:30: node-tools' want moved to a build not yet done, and the node stayed refused after it - // landed until another input moved. A done build of a module the node wants, at its want's or the trunk's build, - // makes a pass, and so does one of a module placed here at its trunk's build though no want names it (novox/hq ADR 0593 - // §6: the composition state holds it at its newest done build before its want moves); a build not done, of another - // build or of a module neither wanted nor placed here, makes none. passes := func() int64 { return composed.Load() } - for _, row := range []struct { - name, module, fp, state string - pass bool - }{ - {"lemurs's wanted build, not done", "lemurs", fp, "building", false}, - {"lemurs's wanted build lands", "lemurs", fp, builds.BuildDone, true}, - {"another build of lemurs", "lemurs", meshcatalogclient.SourcePrefix2 + strings.Repeat("e", 64), builds.BuildDone, false}, - {"a placed module no want names, at its trunk's build", "oldie", meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64), builds.BuildDone, true}, - {"a module neither wanted nor placed here", "ghost", meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64), builds.BuildDone, false}, - } { - before := passes() - key, _ := builds.BuilderBuildKey(row.module, row.fp) - build := builds.BuilderBuild{State: row.state} - if row.state == builds.BuildDone { - build.Manifest = digest - } - raw, err := builds.EncodeBuilderBuild(build) - if err != nil { - t.Fatal(err) - } - b.put(t, builds.BuilderBuildsBucket, key, raw) - quiet(t, out) - if got := passes() > before; got != row.pass { - t.Errorf("%s: a pass made %v, want %v", row.name, got, row.pass) - } - } // Issue 774, 2026-10-31: every put under want.n1.> made a pass, and each pass asked the vault for the node's own // secrets. A key written again with what it held, or a want whose words moved and not its build, makes none. worded, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true, Words: "lemurs is held at aaaaaaaaaaaa"}) @@ -726,7 +698,6 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { {"the node's own want again", buckets.Wants, "want.n1.node", nodeWant, false}, {"lemurs's want with other words", buckets.Wants, "want.n1.lemurs", worded, false}, {"the assignment again", buckets.Assignments, "n1.node", place["n1.node"], false}, - {"lemurs's module again", buckets.Modules, "lemurs", mod, false}, {"lemurs's want at another build", buckets.Wants, "want.n1.lemurs", elsewhere, true}, {"lemurs's want back at its build", buckets.Wants, "want.n1.lemurs", want, true}, }, assignmentRows...) { @@ -747,72 +718,6 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { t.Errorf("%s: a pass made %v, want %v", row.name, got, row.pass) } } - // And today's sequence whole: tool runs at its old build; its want and the trunk move to a build not yet done, and - // the node holds tool, keeping what runs, naming why; the build lands, and with no other change tool is handed at it. - { - tb := newFakeBus() - tool := `{"module":"tool","version":"1","capabilities":["systemd"],"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]},` + - `"resources":[{"id":"service","type":"service","unit":"tool.service","state":"running","boot":"enabled"}]}` - tdigest := meshcatalogclient.ManifestDigest([]byte(tool)) - tman, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "tool", Manifest: tool}) - tkey, _ := meshcatalogclient.ManifestsKey(tdigest) - tb.kv[buckets.Manifests+"/"+tkey] = tman - old, next2 := meshcatalogclient.SourcePrefix2+strings.Repeat("1", 64), meshcatalogclient.SourcePrefix2+strings.Repeat("2", 64) - tenv := tb.env() - var held []string - var hmu sync.Mutex - tenv.Refused = func(why string, h map[string]string) { - hmu.Lock() - defer hmu.Unlock() - held = append(held, why+fmt.Sprint(h)) - } - tenv.Running = func() []compositions.Composed { - return []compositions.Composed{{Node: "n1", Module: "tool", FP: old, Resources: []json.RawMessage{ - json.RawMessage(`{"id":"service","module":"tool","type":"service","unit":"tool.service","state":"running","boot":"enabled"}`)}}} - } - tc := New(tenv) - tout, err := tc.Watch(ctx, "n1.>") - if err != nil { - t.Fatal(err) - } - tplace := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, - Modules: []meshmoduleassigner.PlacedModule{{Module: "tool"}}}) - tmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: next2, Manifest: tdigest, Repository: "novox/t", Commit: "c"}) - twant, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: next2, Self: true}) - building, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: "building"}) - bkey, _ := builds.BuilderBuildKey("tool", next2) - tb.kv[builds.BuilderBuildsBucket+"/"+bkey] = building - tb.putAssigned(t, tplace) - tb.put(t, buckets.Have, "n1.node", have) - tb.put(t, buckets.Modules, "tool", tmod) - tb.put(t, buckets.Wants, "want.n1.tool", twant) - for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, - ResolvedBucket, builds.BuilderBuildsBucket} { - tb.send(bucket, agent.Update{Caught: true}) - } - toolKey, _ := compositions.Key("n1", "tool", next2) - for u := next(t, tout); !u.Caught; u = next(t, tout) { - if u.Key == toolKey { - t.Errorf("tool is handed at a build not yet done") - } - } - hmu.Lock() - if len(held) == 0 || !strings.Contains(held[len(held)-1], "no done build yet") { - t.Errorf("the node was told %q, want tool held for its build not yet done", held) - } - hmu.Unlock() - done, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: tdigest, - Artifacts: []builds.BuiltArtifact{{Name: "files", Kind: "archive", Reference: "artifact-store://tool/files/blobs/sha256:" + strings.Repeat("9", 64), - Digest: "sha256:" + strings.Repeat("9", 64)}}}) - tb.mu.Lock() - tb.kv[builds.BuilderBuildsBucket+"/"+bkey] = done - tb.mu.Unlock() - tb.put(t, builds.BuilderBuildsBucket, bkey, done) - if u := next(t, tout); u.Key != toolKey || u.Deleted { - t.Errorf("the build landed and the node handed %s, want tool at its new build", u.Key) - } - } - // A want removed takes its composition back. removed, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Removed: true}) b.put(t, buckets.Wants, "want.n1.lemurs", removed) @@ -853,10 +758,12 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { b.mu.Lock() b.kv[buckets.Manifests+"/"+kkey] = kman b.mu.Unlock() - kmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"}) + kfp9 := meshcatalogclient.SourcePrefix2 + strings.Repeat("9", 64) + kmod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: kfp9, Manifest: kdigest, Repository: "novox/k", Commit: "c"}) both := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "oldie"}, {Module: "keeper"}}}) - b.put(t, buckets.Modules, "keeper", kmod) + c.takeModule(agent.Update{Key: "keeper", Value: kmod}) + c.holdEntry(t, "keeper", kfp9, keeper) // n1 is held in keys per module and node, which the writes below move. bothPlaced, _ := meshmoduleassignerclient.AssignmentsOf(both) bothNode := bothPlaced["n1"] @@ -869,8 +776,9 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { b.put(t, buckets.Assignments, w.Key, w.Value) } } - b.put(t, buckets.Wants, "want.n1.keeper", want) - keeperKey, _ := compositions.Key("n1", "keeper", fp) + keeperWant, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: kfp9, Self: true}) + b.put(t, buckets.Wants, "want.n1.keeper", keeperWant) + keeperKey, _ := compositions.Key("n1", "keeper", kfp9) var handed string for handed == "" { if u := next(t, out); u.Key == keeperKey { @@ -1138,9 +1046,19 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { return meshcatalog.CatalogEntry{Module: module, Fp: fp, Manifest: m, Repository: "novox/x", Commit: "c", ManifestDigest: meshcatalogclient.ManifestDigest([]byte(body))} } catalogEntries := map[string]meshcatalog.CatalogEntry{fp: entryOfManifest("lemurs", fp, lemurs), kfp: entryOfManifest("keeper", kfp, keeper)} + // A module with an artifact: its build is made from its entry and the parts the candidate gives (design 47 G §628, A-316). + artFp := meshcatalogclient.SourcePrefix2 + strings.Repeat("7", 64) + artManifest := `{"module":"art","version":"1","capabilities":["systemd"],"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]},` + + `"resources":[{"id":"service","type":"service","unit":"art.service","state":"running","boot":"enabled"}]}` + artEntry := entryOfManifest("art", artFp, artManifest) + artEntry.Artifacts = []meshcatalog.Artifact{{Kind: "archive", Language: "none", Ref: "art:files"}} + catalogEntries[artFp] = artEntry senv.Entry = EntryCall(func(_ context.Context, _, verb string, args any) (json.RawMessage, error) { switch in := args.(type) { case meshcatalog.ModuleInput: + if in.Module == "nofp" { // a module the catalog holds and gives no fingerprint (A-312) + return json.Marshal(meshcatalog.ModuleAnswer{Module: meshcatalog.ModuleAnswerModule{Module: "nofp", None: "its build is not made yet"}}) + } for _, e := range catalogEntries { if e.Module == in.Module { return json.Marshal(meshcatalog.ModuleAnswer{Module: meshcatalog.ModuleAnswerModule{Module: e.Module, Repository: e.Repository, Commit: e.Commit, Fp: e.Fp}}) @@ -1237,6 +1155,20 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { candidate("adds", 4, "lemurs", "keeper") candidate("drops", 5) candidate("ghost", 6, "lemurs", "ghost") + candidate("nofp", 11, "lemurs", "nofp") + // A module with an artifact: held when the candidate gives no part for it, built from its entry and the parts when it does. + for group, added := range map[string]map[string]meshmoduleassigner.AddedModule{ + "art-without": nil, + "art-with": {"art": {Version: "1", Parts: map[string]string{"art:files": "sha256:" + strings.Repeat("9", 64)}}}, + } { + p := trunk + p.Modules = []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "art"}} + raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: group, Assignments: &p, Added: added}) + if err != nil { + t.Fatal(err) + } + sb.put(buckets.ProposalCandidates, meshmoduleassigner.CandidatesByProposalNodeKey(group, "n1"), raw, 10) + } // A head deleting n1's assignment file (the seed's revert, mesh-catalog #526): n1 returns to the controller's // compositions, answered as such, not as every module removed. if raw, err := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "deletes"}); err != nil { @@ -1297,11 +1229,22 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { return a.Verdict == nodeengine.CandidateAnswerVerdictDiffers && len(a.Added) == 1 && a.Added[0].Module == "keeper" && len(a.Added[0].Unissued) == 1 && len(a.Refusals) == 0 }, + "art-without": func(a nodeengine.CandidateAnswer) bool { + return a.Verdict == nodeengine.CandidateAnswerVerdictSame && len(a.Held) == 1 && a.Held[0].Module == "art" && + strings.Contains(a.Held[0].Why, "the candidate gives no part for art:files of art's build "+artFp) + }, + "art-with": func(a nodeengine.CandidateAnswer) bool { + return len(a.Added) == 1 && a.Added[0].Module == "art" && len(a.Held) == 0 + }, "drops": func(a nodeengine.CandidateAnswer) bool { return a.Verdict == nodeengine.CandidateAnswerVerdictDiffers && len(a.Removed) == 1 && a.Removed[0] == "lemurs" }, + // A module the verb module answers an error for, or no fingerprint, is held with the verb's reason (A-312), the rest composed. "ghost": func(a nodeengine.CandidateAnswer) bool { - return a.Verdict == nodeengine.CandidateAnswerVerdictRefused && a.Refused != "" + return len(a.Held) == 1 && a.Held[0].Module == "ghost" && strings.Contains(a.Held[0].Why, "the catalog holds no module ghost") && a.Refused == "" + }, + "nofp": func(a nodeengine.CandidateAnswer) bool { + return len(a.Held) == 1 && a.Held[0].Module == "nofp" && strings.Contains(a.Held[0].Why, "gives nofp no fingerprint: its build is not made yet") }, "reach": func(a nodeengine.CandidateAnswer) bool { return a.Verdict == nodeengine.CandidateAnswerVerdictSame && a.Returns == "" && a.Revision == 8 && len(a.Removed) == 0 && @@ -1417,7 +1360,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } k := meshmoduleassigner.CandidatesByProposalNodeKey("same", "n1") sb.del(buckets.ProposalCandidates, k) - waitFor("the deleted group's answer stays", func(a map[string]nodeengine.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 5 }) + waitFor("the deleted group's answer stays", func(a map[string]nodeengine.CandidateAnswer) bool { _, ok := a["same"]; return !ok && len(a) == 8 }) // 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. @@ -1429,7 +1372,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { sb.missing(buckets.ProposalCandidates, false) waitFor("a group deleted while the shadow was down is still answered", func(a map[string]nodeengine.CandidateAnswer) bool { _, ok := a["drops"] - return !ok && len(a) == 4 + return !ok && len(a) == 7 }) // Issue 1126, A7 AC1: a assignment that cannot be read composes nothing and says why; the node never applies a @@ -1477,7 +1420,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { oldRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "older", Assignments: &trunk}) oldKey := meshmoduleassigner.CandidatesByProposalNodeKey("older", "n1") nb.put(buckets.ProposalCandidates, oldKey, oldRaw, 2) - for _, bucket := range []string{buckets.ProposalCandidates, buckets.Assignments, buckets.Have, buckets.Wants, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.ProposalCandidates, buckets.Assignments, buckets.Have, buckets.Wants} { nb.send(bucket, agent.Update{Caught: true}) } select { @@ -1568,10 +1511,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { // Issue 1126, the seed's refusals: what a node composes from its builds and reads of a provider elsewhere. A build // is the builder's key; a templates archive is fetched once per digest and refused when it is not that digest; another // node's served key is read alone, as the node's grant allows it (buscompose Principal.ServedReads). - built, _ := builds.EncodeBuilderBuild(builds.BuilderBuild{State: builds.BuildDone, Manifest: digest}) - buildKey, _ := builds.BuilderBuildKey("lemurs", fp) b.mu.Lock() - b.kv[builds.BuilderBuildsBucket+"/"+buildKey] = built b.kv[ResolvedBucket+"/n2.mailer.domain"] = []byte(`{"value":"example.org","source":"module"}`) b.reads = nil b.mu.Unlock() @@ -1580,7 +1520,14 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { sum := sha256.Sum256(archive) env.Fetch = func(context.Context, string) ([]byte, error) { fetched++; return archive, nil } r := reader{c: New(env)} - if got, err := r.Build(ctx, "lemurs", fp); err != nil || got == nil || got.Manifest != digest { + // A build is made from the entry of its fingerprint that the composer holds, with the parts its want names (design 47 G §421, §628). + var lemursManifest map[string]any + if err := json.Unmarshal([]byte(lemurs), &lemursManifest); err != nil { + t.Fatal(err) + } + r.c.keptEnt = map[string]keptEntry{fp: {entry: meshcatalog.CatalogEntry{Module: "lemurs", Fp: fp, Manifest: lemursManifest, ManifestDigest: digest}}} + r.c.wants["lemurs"] = meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true} + if got, err := r.Build(ctx, "lemurs", fp); err != nil || got == nil || got.State != builds.BuildDone || got.Manifest != digest { t.Errorf("the build of lemurs: %+v, %v", got, err) } if got, err := r.Build(ctx, "lemurs", other); err != nil || got != nil { @@ -1907,6 +1854,15 @@ func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() env := b.env() + afp := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64) + half := catalogHalf{} + half.add(t, "lemurs", fp, lemurs) + half.add(t, "avahi", afp, avahi) + env.Entry = half.call() + state := &fakeComposed{kv: map[string]ComposedEntry{}} + state.keep(t, half, "lemurs", fp) // the entries asked of the catalog half are answered, as kept by an earlier run (§683) + state.keep(t, half, "avahi", afp) + env.ComposedState = state g := &namedGate{gate: &gate{}, seats: []string{"mesh-catalog", "node-engine"}} changed := make(chan struct{}, 1) env.SeatList, env.SeatListChanged = g, changed @@ -1935,11 +1891,13 @@ func TestAnUnreadSeatListHoldsWhatIsNewAndHandsItAtTheRead(t *testing.T) { b.putAssigned(t, place) b.put(t, buckets.Have, "n1.node", have) b.put(t, buckets.Modules, "lemurs", mod) - avahiMod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: avahiDigest, Repository: "novox/x", Commit: "c"}) + avahiMod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: afp, Manifest: avahiDigest, Repository: "novox/x", Commit: "c"}) b.put(t, buckets.Modules, "avahi", avahiMod) b.put(t, buckets.Wants, "want.n1.lemurs", want) + avahiWant, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: afp, Self: true}) + b.put(t, buckets.Wants, "want.n1.avahi", avahiWant) b.put(t, buckets.Wants, "want.n1.node", nodeWant) - for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } handed := map[string]bool{} @@ -2020,6 +1978,9 @@ func TestAHeldOwnCompositionThatWasHandedIsNotDropped(t *testing.T) { ctx, cancel := context.WithCancel(context.Background()) defer cancel() env := b.env() + half := catalogHalf{} + half.add(t, "lemurs", fp, lemurs) + env.Entry = half.call() g := &gate{} changed := make(chan struct{}, 1) env.SeatList, env.SeatListChanged = g, changed @@ -2041,7 +2002,7 @@ func TestAHeldOwnCompositionThatWasHandedIsNotDropped(t *testing.T) { b.put(t, buckets.Modules, "lemurs", mod) b.put(t, buckets.Wants, "want.n1.lemurs", want) b.put(t, buckets.Wants, "want.n1.node", nodeWant) - for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } // The list is read: the node's own composition is handed. Then the reader cannot act again (it restarted and has @@ -2151,7 +2112,7 @@ func TestTheHolderOfTheCatalogIsStartedFromTheOwnCompositionBeforeTheListIsReadA mk, _ := meshcatalogclient.ManifestsKey(digest) b.kv[buckets.Manifests+"/"+mk] = man rec, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64), Manifest: digest, Repository: "novox/x", Commit: "c"}) - b.put(t, buckets.Modules, m, rec) + c.takeModule(agent.Update{Key: m, Value: rec}) // a module run inside the node's own composition: no want names it, so no entry is asked hk, _ := nodeengineclient.HaveKey("n1", m) hv, _ := nodeengineclient.EncodeHave(nodeengine.HaveByNodeModuleValue{Build: "build-1", Phase: nodeengineclient.PhaseRunning, Health: nodeengineclient.HealthOK}) b.put(t, buckets.Have, hk, hv) @@ -2163,7 +2124,7 @@ func TestTheHolderOfTheCatalogIsStartedFromTheOwnCompositionBeforeTheListIsReadA b.putAssigned(t, place) b.put(t, buckets.Have, "n1.node", have) b.put(t, buckets.Wants, "want.n1.node", nodeWant) - for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } servicesIn := func(u agent.Update) []string { diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index e272cf8c..1fff9c71 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -16,6 +16,7 @@ import ( engcomp "github.com/novox/mesh-host/internal/compositions" "github.com/novox/mesh-host/internal/judge" "maps" + "slices" "sort" "strings" "sync" @@ -592,9 +593,16 @@ func (s *Shadow) answerProposal(ctx context.Context, proposal string) (nodeengin inOwn := c.inOwnOf(*assignment) c.mu.Unlock() fps := shadowBuilds(running, *assignment, modules, inOwn) + if why := c.askBuildsEntries(fps); why != "" { + // Its answer wakes the shadow (entryMoved), which composes this proposal again. + c.missed.Add(1) + return nodeengine.CandidateAnswer{Proposal: proposal, Revision: int64(rev), Verdict: nodeengine.CandidateAnswerVerdictRefused, + Refused: why + ", so this answer waits for it"}, true + } // A module the candidate adds that no want here names: the trunk's build is asked of the catalog half's verb module, and its // entry by the verb entry (design 47 G §421, A-312). One with no build to take is held, with the reason. heldAdded := map[string]string{} + addedWants := map[string]meshwants.WantsByWantNodeModuleValue{} folded := map[string]bool{} for _, m := range inOwn { folded[m] = true @@ -620,6 +628,13 @@ func (s *Shadow) answerProposal(ctx context.Context, proposal string) (nodeengin heldAdded[pm.Module] = fmt.Sprintf("the entry of %s's build %s holds no manifest that can be read: %v", pm.Module, fp, err) continue } + // The module is built from its entry and the parts the candidate gives (A-316); an artifact it gives no part for holds it. + w := meshwants.WantsByWantNodeModuleValue{Fp: fp, Parts: own.c.Added[pm.Module].Parts, Version: own.c.Added[pm.Module].Version} + if missing := partsMissing(entry, w); missing != "" { + heldAdded[pm.Module] = fmt.Sprintf("the candidate gives no part for %s of %s's build %s, so the node composes nothing new of it", missing, pm.Module, fp) + continue + } + addedWants[pm.Module] = w m := meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: entry.Repository, Commit: entry.Commit, Path: entry.Path, Needs: entry.Needs, Packages: entry.Packages, Verdict: entry.Verdict} if entry.Inputs != nil { @@ -642,14 +657,41 @@ func (s *Shadow) answerProposal(ctx context.Context, proposal string) (nodeengin // 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]meshwants.WantsByWantNodeModuleValue{agent.NodeModule: {}} + c.mu.Lock() for module := range fps { held[module] = meshwants.WantsByWantNodeModuleValue{Fp: fps[module]} + if w, ok := c.wants[module]; ok && w.Fp == fps[module] { + held[module] = w // the want's parts give the build's artifacts their digests (buildOf) + } + if w, ok := addedWants[module]; ok { + held[module] = w // so do the parts the candidate gives a module it adds + } } - c.mu.Lock() c.wants = held c.mu.Unlock() + // A module the candidate adds that cannot be built is held with the reason, and the rest is composed without it: composed with it, the + // assignment would be refused for a module nothing provides. + if len(heldAdded) > 0 { + without := *assignment + without.Modules = nil + for _, pm := range assignment.Modules { + if _, held := heldAdded[pm.Module]; !held { + without.Modules = append(without.Modules, pm) + } + } + given[node] = &without + } a := frombus.Answer(ctx, reader{c: c}, c.env.Mesh, frombus.Scope{Node: node, KeptAt: kept, Assignments: given, InOwn: inOwn, SameValue: c.sameValue, HeldToday: s.heldToday(heldAdded)}, fps, running, proposal, rev) + for _, module := range slices.Sorted(maps.Keys(heldAdded)) { + a.Held = append(a.Held, nodeengine.CandidateRefusal{Module: module, Why: heldAdded[module]}) + } + // What this proposal added is its own: another proposal that places the module without giving its parts must not find it composed. + c.mu.Lock() + for module := range addedWants { + delete(c.modules, module) + } + c.mu.Unlock() s.mu.Lock() ran := make(map[string]string, len(s.ran)) for m, d := range s.ran {