diff --git a/cmd/mesh-host/nodeagent.go b/cmd/mesh-host/nodeagent.go index 098c8753..9a10734f 100644 --- a/cmd/mesh-host/nodeagent.go +++ b/cmd/mesh-host/nodeagent.go @@ -1135,6 +1135,9 @@ 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) } diff --git a/cmd/mesh-host/nodeagent_integration_test.go b/cmd/mesh-host/nodeagent_integration_test.go index fc86409e..4d3f4021 100644 --- a/cmd/mesh-host/nodeagent_integration_test.go +++ b/cmd/mesh-host/nodeagent_integration_test.go @@ -79,6 +79,55 @@ type itBed struct { // holder re-serves (issue 1966). seatAway atomic.Bool findings nats.KeyValue + // entries are the entries the stand-in of the seat mesh-catalog answers its verb entry from, by fingerprint (design 47 G §421). + entries map[string]meshcatalog.CatalogEntry +} + +// registerEntry makes the stand-in answer the entry of fp: module's manifest and its artifacts (refs `:`). +func (b *itBed) registerEntry(module, fp, manifest string, artifacts ...string) { + b.t.Helper() + var m map[string]any + if err := json.Unmarshal([]byte(manifest), &m); err != nil { + b.t.Fatal(err) + } + e := meshcatalog.CatalogEntry{Module: module, Fp: fp, Manifest: m, ManifestDigest: meshcatalogclient.ManifestDigest([]byte(manifest)), + Repository: "novox/" + module, Commit: strings.Repeat("1", 40)} + for _, a := range artifacts { + e.Artifacts = append(e.Artifacts, meshcatalog.Artifact{Ref: module + ":" + a, Kind: "archive", Language: "go"}) + } + b.mu.Lock() + b.entries[fp] = e + b.mu.Unlock() +} + +// running says the seat node-engine's event running of module on node at fp, as that node's engine does. +func (b *itBed) running(node, module, fp, manifestDigest string) { + b.t.Helper() + raw, _ := json.Marshal(nodeengine.Running{V: 1, Node: node, Module: module, Fingerprint: fp, Manifest: manifestDigest, At: time.Now().UTC()}) + if _, err := b.js.Publish(nodeengine.SubjectRunning(node, module), raw); err != nil { + b.t.Fatal(err) + } +} + +// standInEntry answers the seat mesh-catalog's verb entry from the registered entries, and as the holder refuses an unknown one. +func (b *itBed) standInEntry() { + sub, err := b.conn.Subscribe("mesh.seat.mesh-catalog.call.entry.*", func(m *nats.Msg) { + var in meshcatalog.EntryInput + _ = json.Unmarshal(m.Data, &in) + b.mu.Lock() + e, ok := b.entries[in.Fingerprint] + b.mu.Unlock() + if !ok { + _ = m.Respond([]byte(`{"error":"no entry of fingerprint ` + in.Fingerprint + ` is kept"}`)) + return + } + out, _ := json.Marshal(map[string]any{"result": meshcatalog.EntryAnswer{Entry: e}}) + _ = m.Respond(out) + }) + if err != nil { + b.t.Fatal(err) + } + b.t.Cleanup(func() { _ = sub.Unsubscribe() }) } // aServer is an embedded bus with JetStream and two users: the test's, unrestricted, and the node's, granted exactly what @@ -131,7 +180,7 @@ func newItBedAs(t *testing.T, perms nodegrant.Permissions) *itBed { builtFor = "arch" t.Cleanup(func() { builtFor = was }) - b := &itBed{t: t, dir: t.TempDir()} + b := &itBed{t: t, dir: t.TempDir(), entries: map[string]meshcatalog.CatalogEntry{}} b.conn, b.nodeConn = aServer(t, perms, func(err error) { b.mu.Lock() b.refusals = append(b.refusals, err.Error()) @@ -167,7 +216,13 @@ func newItBedAs(t *testing.T, perms nodegrant.Permissions) *itBed { t.Fatal(err) } } + // The stream the seat node-engine's event running is kept in, the last of each subject (design 47 G §199). + if _, err = b.js.AddStream(&nats.StreamConfig{Name: nodeengine.StreamRunning, Subjects: []string{nodeengine.SubjectRunningAll}, + MaxMsgsPerSubject: 1}); err != nil { + t.Fatal(err) + } b.standInSeat() + b.standInEntry() b.opts = options{state: filepath.Join(b.dir, "state", "state.json"), timeout: 2 * time.Second} if err := os.MkdirAll(filepath.Dir(b.opts.state), 0o755); err != nil { t.Fatal(err) @@ -435,8 +490,7 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { 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"},` + - `{"id":"tools","type":"package","package":"web-tools"}],` + - `"build":{"artifacts":[{"name":"files","kind":"archive","from":"files"}]}}` + `{"id":"tools","type":"package","package":"web-tools"}]}` digest := meshcatalogclient.ManifestDigest([]byte(man)) mraw, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "web", Manifest: man}) mkey, _ := meshcatalogclient.ManifestsKey(digest) @@ -464,9 +518,6 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { 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 - if checkBuild { - firstBuild = []byte(`{"v":1,"state":"no such state"}`) - } for _, w := range []struct { kv nats.KeyValue key string @@ -479,6 +530,13 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { t.Fatal(err) } } + // Design 47 G §421: the node reads the entry its want names, through the seat mesh-catalog's verb entry, and the part's digest + // from the want; what it hears another node run (the mailer on it2) is read by the same fingerprint. + if !checkBuild { + b.registerEntry("web", fp, man) + } + b.registerEntry("mailer", fpIt("b"), mailer) + b.running("it2", "mailer", fpIt("b"), mdigest) want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true}) wkey := meshwants.WantsByWantNodeModuleKey(itNode, "web") if _, err := b.wants.Put(wkey, want); err != nil { @@ -499,17 +557,19 @@ func h8(t *testing.T, b *itBed, checkBuild bool) (func(), h8state) { t.Fatal(err) } if checkBuild { - // The build of the placed module is read through the node's session: one that does not decode is said, and nothing - // is applied; a done one, then a config event, and the node composes. - within(t, 5*time.Second, "the node did not read web's build key through its session", func() bool { - b.mu.Lock() - defer b.mu.Unlock() - return strings.Contains(strings.Join(b.said, "\n"), "the builder's "+buildKey) + _ = 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. + within(t, 5*time.Second, "the have does not say the entry of web's build is missing", func() bool { + h := b.have("web") + return h.Phase == nec.PhaseHeld && strings.HasPrefix(h.Detail, agent.EntryMissing) && strings.Contains(h.Detail, fp) }) if b.file("web") != "" { - t.Fatal("composed with a build key that does not decode") + t.Fatal("composed with an entry the catalog half does not hold") } - if _, err := builds.Put(buildKey, done); err != nil { + b.registerEntry("web", fp, man) + if err := composerNow.Load().EntryAgain(context.Background(), fp); err != nil { t.Fatal(err) } if _, err := resolved.Put(itNode+".web.greeting", []byte(`{"value":"bonjour","source":"module"}`)); err != nil { @@ -799,6 +859,7 @@ func TestNatsTheBusServerIsTakenOverAndReplacedOnlyWhenItsBuildChangesIt(t *test mraw, _ := meshcatalogclient.EncodeManifest(meshcatalog.ManifestsByDigestValue{Module: "nats", Manifest: man}) mkey, _ := meshcatalogclient.ManifestsKey(digest) modv, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/mesh-catalog", Commit: strings.Repeat("3", 40)}) + b.registerEntry("nats", fp, man) // the node reads the entry its want names (design 47 G §421) if _, err := manifests.Put(mkey, mraw); err != nil { t.Fatal(err) } diff --git a/internal/agent/agent.go b/internal/agent/agent.go index 2b40d698..4e288c90 100644 --- a/internal/agent/agent.go +++ b/internal/agent/agent.go @@ -508,6 +508,9 @@ const NotComposedDetail = "the node cannot compose it, and keeps running what it // composerMark says h is a mark NotComposed wrote: failed, its detail opening with NotComposedDetail. func composerMark(h Have) bool { + if h.Phase == string(nodeengineclient.PhaseHeld) && heldForEntry(h.Detail) { + return true + } return h.Phase == string(nodeengineclient.PhaseFailed) && strings.HasPrefix(h.Detail, NotComposedDetail) } @@ -651,8 +654,14 @@ func (a *Agent) NotComposedEach(ctx context.Context, why string, held map[string a.mu.Lock() a.notComposed[module] = mark a.mu.Unlock() + phase, detail := string(nodeengineclient.PhaseFailed), NotComposedDetail+why + if heldForEntry(why) { + // The entry its want names is not held: the phase held, its detail opening entry-asked: or entry-missing: and + // naming the fingerprint (design 47 G section 421). + phase, detail = string(nodeengineclient.PhaseHeld), why + } a.writeHave(ctx, Have{Module: module, FP: mark.prior.FP, Composed: mark.prior.Composed, Sealed: mark.prior.Sealed, - Phase: string(nodeengineclient.PhaseFailed), Detail: NotComposedDetail + why}) + Phase: phase, Detail: detail}) } } diff --git a/internal/agent/agent_test.go b/internal/agent/agent_test.go index a609f558..34268265 100644 --- a/internal/agent/agent_test.go +++ b/internal/agent/agent_test.go @@ -1772,6 +1772,12 @@ func TestTheNodesOwnCompositionIsAppliedWhenItDiffersAndARecordMoveWaitsForAYes( if h := b.haves.last("web"); h.Phase != string(nodeengineclient.PhaseRunning) || h.Composed != dig { t.Errorf("web, which composes, is marked beside idle: %+v", h) } + // Design 47 G §421: a module whose want's entry the catalog half holds none of is the phase held, its detail opening + // entry-missing: and naming the fingerprint; composed again, its have is what it runs. + b.ag.NotComposedEach(context.Background(), "", map[string]string{"idle": EntryMissing + " the catalog half holds no entry of the build src2:x"}) + if h := b.haves.last("idle"); h.Phase != string(nodeengineclient.PhaseHeld) || h.Composed != idleDig || h.Detail != EntryMissing+" the catalog half holds no entry of the build src2:x" { + t.Errorf("idle's have while its entry is missing: %+v", h) + } b.ag.NotComposedEach(context.Background(), "", nil) if h := b.haves.last("idle"); h.Phase != string(nodeengineclient.PhaseRunning) || h.Composed != idleDig { t.Errorf("idle's have once it composes again: %+v", h) diff --git a/internal/agent/parts.go b/internal/agent/parts.go index bb5a1f83..b324b9a7 100644 --- a/internal/agent/parts.go +++ b/internal/agent/parts.go @@ -13,6 +13,19 @@ import ( // disagree (design 47 G section 524). const PartsDiffer = "parts-differ:" +// EntryAsked and EntryMissing open the detail of a have held for the entry the want names (design 47 G section 421 and +// section 614): the entry is asked of the catalog half and not yet answered, or the half answered that it holds none. Both +// are the phase held, and each names the fingerprint. +const ( + EntryAsked = "entry-asked:" + EntryMissing = "entry-missing:" +) + +// heldForEntry says whether why is a hold for the entry a want names. +func heldForEntry(why string) bool { + return strings.HasPrefix(why, EntryAsked) || strings.HasPrefix(why, EntryMissing) +} + // partsDifference checks a want's `parts` against the digests of the archive, process and container resources of the // composition, which are the digests the apply fetches and checks (design 47 G section 524). A want with no `parts` (an // assigner of before G, or a node whose record does not state `want-parts`) is applied as today: it returns "". With diff --git a/internal/link/running.go b/internal/link/running.go index 722ce19a..fa24b0d1 100644 --- a/internal/link/running.go +++ b/internal/link/running.go @@ -93,3 +93,15 @@ func (q *Session) FollowRunning(ctx context.Context, heard func(nodeengine.Runni <-ctx.Done() return nil } + +// OverNATS reads the stream as the link's session does, over its connection. +func (b OverNATS) FirstRunning(ctx context.Context, first func([]byte)) error { + return (&natsLink{js: b.JS}).FirstRunning(ctx, first) +} + +// HearRunning follows the stream over the connection (see natsLink.HearRunning). +func (b OverNATS) HearRunning(consumer string, heard func([]byte)) (func(), error) { + return (&natsLink{js: b.JS}).HearRunning(consumer, heard) +} + +var _ RunningBus = OverNATS{} diff --git a/internal/nodegrant/testdata/node.json b/internal/nodegrant/testdata/node.json index 6e82524d..0270797c 100644 --- a/internal/nodegrant/testdata/node.json +++ b/internal/nodegrant/testdata/node.json @@ -21,7 +21,9 @@ "$JS.API.CONSUMER.CREATE.KV_node-engine_have.*.$KV.node-engine_have.NODE.>", "$JS.API.CONSUMER.CREATE.OBJ_declaration-files-NODE.*.$O.declaration-files-NODE.C.*", "$JS.API.CONSUMER.CREATE.SEAT_mesh-conditions_findings-asked", + "$JS.API.CONSUMER.CREATE.SEAT_node-engine_running", "$JS.API.CONSUMER.CREATE.SEAT_mesh-conditions_findings-asked.>", + "$JS.API.CONSUMER.CREATE.SEAT_node-engine_running.>", "$JS.API.CONSUMER.DELETE.KV_composed-NODE.>", "$JS.API.CONSUMER.DELETE.KV_mesh-config_resolved.>", "$JS.API.CONSUMER.DELETE.KV_mesh-delivery-assigner_assignments.>", @@ -41,6 +43,8 @@ "$JS.API.CONSUMER.DELETE.KV_node-engine_have.>", "$JS.API.CONSUMER.DELETE.OBJ_declaration-files-NODE.*", "$JS.API.CONSUMER.DELETE.SEAT_mesh-conditions_findings-asked.>", + "$JS.API.CONSUMER.DELETE.SEAT_node-engine_running.>", + "$JS.API.CONSUMER.INFO.SEAT_node-engine_running.>", "$JS.API.CONSUMER.INFO.NODES.NODE", "$JS.API.DIRECT.GET.KV_composed-NODE.>", "$JS.API.DIRECT.GET.KV_mesh-config_resolved.$KV.mesh-config_resolved.NODE.>", @@ -79,6 +83,7 @@ "$JS.API.STREAM.INFO.KV_mesh-provisioner_provisions", "$JS.API.STREAM.INFO.KV_node-engine_have", "$JS.API.STREAM.INFO.SEAT_mesh-conditions_findings-asked", + "$JS.API.STREAM.INFO.SEAT_node-engine_running", "$JS.FC.KV_composed-NODE.>", "$JS.FC.KV_mesh-config_resolved.>", "$JS.FC.KV_mesh-delivery-assigner_assignments.>", @@ -110,6 +115,7 @@ "$SRV.PING.>", "$SRV.PING.node-tools.NODE", "mesh.control.NODE.>", + "mesh.seat.mesh-catalog.call.entry.node~NODE", "mesh.seat.mesh-catalog.call.seat-definitions.node~NODE", "mesh.seat.node-backup.call.backed-up.NODE.node~NODE", "mesh.seat.node-backup.call.now.NODE.node~NODE", diff --git a/internal/selfcompose/composedstate.go b/internal/selfcompose/composedstate.go index aef6dfbc..34c0f2c9 100644 --- a/internal/selfcompose/composedstate.go +++ b/internal/selfcompose/composedstate.go @@ -86,6 +86,11 @@ func (c *Composer) publishComposed(ctx context.Context) { // the indexer's, where the builder holds a done build of it; else the want's; else what runs of it. A module with none // is left out. func (c *Composer) newestFPs(ctx context.Context, at *published) map[string]string { + if c.byEntries() { + // The newest build is the indexer's (and the builder's done key says it is built): neither is read, and the node + // follows no branch (design 47 G §421, §628), so the state holds the wanted builds alone. + return maps.Clone(at.fps) + } out := map[string]string{} r := reader{c: c} for _, module := range at.placed { diff --git a/internal/selfcompose/entries.go b/internal/selfcompose/entries.go new file mode 100644 index 00000000..6b00c5b1 --- /dev/null +++ b/internal/selfcompose/entries.go @@ -0,0 +1,516 @@ +package selfcompose + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "reflect" + "sort" + "strings" + "time" + + "git.novox.be/novox/mesh-delivery/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + + "github.com/novox/mesh-host/internal/agent" +) + +// **The catalog half, read by fingerprint** (design 47 G §421, §683, §698; the lead's answers A-47 and A-50). The node-engine +// follows no branch: for each build its wants and the runnings it heard name, it holds the catalog graph's entry of that +// fingerprint, asked of the seat mesh-catalog's verb `entry` once (meshcatalogclient.Entries) and kept in its composition +// state `composed-` beside the compositions (`entry.`, the entry as `entry` answers it), with each want it +// composes from (`want.`), written whole before the composition that uses them and read at its start before it asks +// anything. A module's manifest is the entry's, and each artifact's reference is made from the digest its want's parts name +// (referenceOf), so it reads neither the indexer's states nor the builder's. + +// The keys of the kept entries and wants in the composition state. +const ( + keptEntryPrefix = "entry." + keptWantPrefix = "want." + // keptWithin bounds one write of the composition state from an answer of the catalog half. + keptWithin = 15 * time.Second +) + +// keptWant is a want as the composition state keeps it (§683): the want beside when this node first kept it. +type keptWant struct { + V int `json:"v"` + Since time.Time `json:"since"` + Want meshwants.WantsByWantNodeModuleValue `json:"want"` +} + +// keptEntry is an entry held from the composition state or an answer, with when it was kept. +type keptEntry struct { + entry meshcatalog.CatalogEntry + since time.Time +} + +// EntryRow is one entry the node-engine keeps, as the seat's verb first-views lists it (§698). +type EntryRow struct { + Fingerprint string `json:"fingerprint"` + State string `json:"state"` + Since time.Time `json:"since"` +} + +// WantRow is one want the node-engine keeps, as the seat's verb first-views lists it (§698). +type WantRow struct { + Module string `json:"module"` + Fingerprint string `json:"fingerprint"` + KeptSince time.Time `json:"kept-since"` +} + +// byEntries says the composer reads the catalog half by fingerprint (the node's own composer), as the shadow, which keeps +// its reads of the indexer until it goes with the checker (§628), does not. +func (c *Composer) byEntries() bool { return c.entries != nil } + +// newEntries makes the composer's cache of entries over env's call of `entry`. +func (c *Composer) newEntries() { + if c.env.Entry == nil { + return + } + c.entries = meshcatalogclient.NewEntries(c.env.Entry) + c.entries.Now = c.now + c.entries.Changed = c.entryChanged + c.keptEnt = map[string]keptEntry{} + c.keptWants = map[string]bool{} + c.keptWantVal = map[string]keptWant{} +} + +// entryKey is an entry's key in the composition state, wantKey a want's. +// +// The fingerprint is keyed by its digits, as the compositions' keys are (engcomp.Key): a key holds no colon. +func entryKey(fp string) string { + return keptEntryPrefix + strings.TrimPrefix(fp, meshcatalogclient.SourcePrefix2) +} + +func wantStateKey(module string) string { return keptWantPrefix + module } + +// loadKept reads the entries and wants the composition state keeps, once at the start, before anything is asked of the catalog +// half (§683). A want read stands only until the live read of the wants answers (endKeptWants). A state that cannot be read +// is said, and what it would have kept is asked of `entry` and read from the wants as they come. +func (c *Composer) loadKept(ctx context.Context) { + if !c.byEntries() || c.env.ComposedState == nil { + return + } + kept, err := c.env.ComposedState.Entries(ctx) + if err != nil { + c.sayOnceEach("the entries and wants this node kept cannot be read, so each entry its wants name is asked of the catalog half: " + err.Error()) + return + } + now := c.now() + c.mu.Lock() + defer c.mu.Unlock() + for key, e := range kept { + switch { + case strings.HasPrefix(key, keptEntryPrefix): + var a meshcatalog.EntryAnswer + if json.Unmarshal(e.Value, &a) != nil || a.Entry.Module == "" { + c.env.Say("the kept entry " + key + " cannot be read, and is asked of the catalog half again") + continue + } + c.keptEnt[meshcatalogclient.SourcePrefix2+strings.TrimPrefix(key, keptEntryPrefix)] = keptEntry{entry: a.Entry, since: now} + case strings.HasPrefix(key, keptWantPrefix): + var w keptWant + module := strings.TrimPrefix(key, keptWantPrefix) + if json.Unmarshal(e.Value, &w) != nil || w.Want.Fp == "" { + c.env.Say("the kept want " + key + " cannot be read, so it stands until the wants are read") + continue + } + if _, live := c.wants[module]; !live { + c.wants[module] = w.Want + c.keptWants[module] = true + } + c.keptWantVal[module] = w + } + } +} + +// endKeptWants ends the standing of the kept wants once the live read of the wants answered: from then the wants the seat +// says are the ones the node composes from, never a want older than its stream's (§683). A kept want the read did not +// announce is gone. +func (c *Composer) endKeptWants() { + if !c.byEntries() { + return + } + c.mu.Lock() + defer c.mu.Unlock() + for module := range c.keptWants { + delete(c.wants, module) + delete(c.keptWants, module) + } +} + +// entryChanged is told of each entry answered or found missing: an answer is kept in the composition state before the pass it +// asks (so the composition that uses it finds it kept), and a pass is asked. +func (c *Composer) entryChanged(row meshcatalogclient.EntryRow) { + if row.State == meshcatalogclient.EntryAnswered && row.Entry != nil { + c.mu.Lock() + c.keptEnt[row.Fingerprint] = keptEntry{entry: *row.Entry, since: row.Since} + c.mu.Unlock() + c.keep(entryKey(row.Fingerprint), meshcatalog.EntryAnswer{Entry: *row.Entry}) + } + c.askRehand() +} + +// keep writes one value into the composition state; a state that cannot be written is said, and the value is kept again by +// the next pass. +func (c *Composer) keep(key string, v any) { + if c.env.ComposedState == nil { + return + } + raw, err := json.Marshal(v) + if err != nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), keptWithin) + defer cancel() + c.keptWrite.Lock() + defer c.keptWrite.Unlock() + if _, err := c.env.ComposedState.Put(ctx, key, raw); err != nil { + c.sayOnceEach("the node's composition state cannot keep " + key + ": " + err.Error()) + } +} + +// forget deletes one key from the composition state. +func (c *Composer) forget(key string) { + if c.env.ComposedState == nil { + return + } + ctx, cancel := context.WithTimeout(context.Background(), keptWithin) + defer cancel() + c.keptWrite.Lock() + defer c.keptWrite.Unlock() + if err := c.env.ComposedState.Delete(ctx, key); err != nil { + c.sayOnceEach("the node's composition state cannot drop " + key + ": " + err.Error()) + } +} + +// entryOf is the entry of fp as the composer holds it: kept, or asked of the catalog half once (Entries.Need). +func (c *Composer) entryOf(fp string) (meshcatalog.CatalogEntry, meshcatalogclient.EntryState) { + c.mu.Lock() + k, ok := c.keptEnt[fp] + c.mu.Unlock() + if ok { + return k.entry, meshcatalogclient.EntryAnswered + } + return c.entries.Need(context.Background(), fp) +} + +// wantedBuilds is each fingerprint a want of this node or a running heard names, of the new form, with whether a want names it. +func (c *Composer) wantedBuilds() map[string]bool { + out := map[string]bool{} + c.mu.Lock() + defer c.mu.Unlock() + for module, w := range c.wants { + if module != agent.NodeModule && strings.HasPrefix(w.Fp, meshcatalogclient.SourcePrefix2) { + out[w.Fp] = true + } + } + for _, mods := range c.heard { + for _, run := range mods { + if !run.stopped && strings.HasPrefix(run.fingerprint, meshcatalogclient.SourcePrefix2) { + if _, named := out[run.fingerprint]; !named { + out[run.fingerprint] = false + } + } + } + } + return out +} + +// syncEntries asks the entry of each build a want or a running names, drops the one nothing names any more, keeps each want it +// composes from, and lays the picture frombus reads (c.modules, c.manifests) from the entries it holds. It runs at the start of +// each pass. +func (c *Composer) syncEntries(ctx context.Context) { + if !c.byEntries() { + return + } + need := c.wantedBuilds() + for fp := range need { + c.entryOf(fp) + } + for _, row := range c.entries.Rows() { + if _, named := need[row.Fingerprint]; !named { + c.entries.Release(row.Fingerprint) + } + } + c.mu.Lock() + var drop []string + for fp := range c.keptEnt { + if _, named := need[fp]; !named { + delete(c.keptEnt, fp) + drop = append(drop, entryKey(fp)) + } + } + type pending struct { + module string + kw keptWant + } + var put []pending + var gone []string + now := c.now() + for module, w := range c.wants { + if module == agent.NodeModule || c.keptWants[module] { + continue + } + had, ok := c.keptWantVal[module] + if ok && reflect.DeepEqual(had.Want, w) { + continue + } + since := now + if ok && had.Want.Fp == w.Fp { + since = had.Since + } + kw := keptWant{V: 1, Since: since, Want: w} + c.keptWantVal[module] = kw + put = append(put, pending{module, kw}) + } + for module := range c.keptWantVal { + if _, held := c.wants[module]; !held { + delete(c.keptWantVal, module) + gone = append(gone, wantStateKey(module)) + } + } + c.mu.Unlock() + // Written whole before the composition that uses them (§683). + for _, p := range put { + c.keep(wantStateKey(p.module), p.kw) + } + for _, key := range append(drop, gone...) { + c.forget(key) + } + c.layModules() +} + +// layModules lays c.modules and c.manifests from the entries held: what each node was heard to run first, then each module this +// node's want names, at the want's fingerprint (a want wins). +func (c *Composer) layModules() { + c.mu.Lock() + defer c.mu.Unlock() + modules := map[string]meshcatalog.ModulesByModuleValue{} + lay := func(module, fp string) { + k, ok := c.keptEnt[fp] + if !ok || k.entry.Module == "" { + return + } + e := k.entry + digest, raw, err := entryManifest(e) + if err != nil { + return + } + m := meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: e.Repository, Commit: e.Commit, Path: e.Path, + Needs: e.Needs, Packages: e.Packages, Verdict: e.Verdict} + if e.Inputs != nil { + m.Problems, m.Warnings, m.Unknown = e.Inputs.Problems, e.Inputs.Warnings, e.Inputs.Unknown + } + modules[module] = m + c.manifests[digest] = raw + } + nodes := make([]string, 0, len(c.heard)) + for node := range c.heard { + nodes = append(nodes, node) + } + sort.Strings(nodes) + for _, node := range nodes { + for module, run := range c.heard[node] { + if !run.stopped && run.fingerprint != "" { + lay(module, run.fingerprint) + } + } + } + for module, w := range c.wants { + if module != agent.NodeModule && w.Fp != "" { + lay(module, w.Fp) + } + } + // A module an assignment places whose entry is not held (not wanted here, not heard running, or its entry asked and not + // answered) stands in the catalogue as its name alone, so the placement still finds it: a module held for its entry offers + // nothing to the rest until its entry is held (§421). + for _, p := range c.assignments { + for _, pm := range p.Modules { + if _, held := modules[pm.Module]; held || pm.Module == "" { + continue + } + raw, _ := json.Marshal(map[string]any{"module": pm.Module, "version": "0"}) + sum := sha256.Sum256(raw) + digest := "sha256:" + hex.EncodeToString(sum[:]) + // At the fingerprint its want names, so it is no build behind the catalog's and its hold (entryHolds) is what says it. + modules[pm.Module] = meshcatalog.ModulesByModuleValue{Manifest: digest, Fp2: c.wants[pm.Module].Fp} + c.manifests[digest] = raw + } + } + c.modules = modules +} + +// entryManifest is an entry's manifest as a reader takes it, with its digest: the one the indexer set at take-in, or, where it set +// none, the digest of the manifest's bytes. +func entryManifest(e meshcatalog.CatalogEntry) (digest string, raw []byte, err error) { + if raw, err = json.Marshal(e.Manifest); err != nil { + return "", nil, err + } + digest = e.ManifestDigest + if digest == "" { + sum := sha256.Sum256(raw) + digest = "sha256:" + hex.EncodeToString(sum[:]) + } + return digest, raw, nil +} + +// entryHolds is, of each module this node's want names, why its entry is not held: asked and not answered, or answered not +// found. It is the phase held with a detail opening entry-asked: or entry-missing: and naming the fingerprint (§421, §614). +func (c *Composer) entryHolds() map[string]string { + if !c.byEntries() { + return nil + } + c.mu.Lock() + wants := map[string]meshwants.WantsByWantNodeModuleValue{} + for module, w := range c.wants { + if module != agent.NodeModule { + wants[module] = w + } + } + c.mu.Unlock() + out := map[string]string{} + for module, w := range wants { + if !strings.HasPrefix(w.Fp, meshcatalogclient.SourcePrefix2) { + continue + } + switch e, state := c.entryOf(w.Fp); state { + case meshcatalogclient.EntryAsked: + out[module] = fmt.Sprintf("%s the entry of the build %s is asked of the catalog half and not yet answered, so the node composes nothing new of %s", + agent.EntryAsked, w.Fp, module) + case meshcatalogclient.EntryNotFound: + out[module] = fmt.Sprintf("%s the catalog half holds no entry of the build %s, so the node composes nothing new of %s", + agent.EntryMissing, w.Fp, module) + default: + if missing := partsMissing(e, w); missing != "" { + out[module] = fmt.Sprintf("the want of %s names no digest for %s, so the node composes nothing new of it", module, missing) + } + } + } + return out +} + +// partsMissing names the artifacts of e whose digest w's parts do not give; "" when each has one. +func partsMissing(e meshcatalog.CatalogEntry, w meshwants.WantsByWantNodeModuleValue) string { + var out []string + for _, a := range e.Artifacts { + if w.Parts[a.Ref] == "" { + out = append(out, a.Ref) + } + } + sort.Strings(out) + return strings.Join(out, ", ") +} + +// referenceOf is an artifact's reference made from the digest its want's part names, in the form the artifact store is +// addressed by (the lead's answer A-50): artifact-store:////blobs/. +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. +func (c *Composer) buildOf(module, fp string) *builds.BuilderBuild { + c.mu.Lock() + w, wanted := c.wants[module] + k, ok := c.keptEnt[fp] + c.mu.Unlock() + if !wanted || w.Fp != fp || !ok || partsMissing(k.entry, w) != "" { + return nil + } + digest, raw, err := entryManifest(k.entry) + if err != nil { + return nil + } + c.mu.Lock() + c.manifests[digest] = raw + c.mu.Unlock() + b := builds.BuilderBuild{State: builds.BuildDone, Manifest: digest} + for _, a := range k.entry.Artifacts { + _, name, _ := strings.Cut(a.Ref, ":") + digest := w.Parts[a.Ref] + b.Artifacts = append(b.Artifacts, builds.BuiltArtifact{Name: name, Kind: a.Kind, Digest: digest, + Reference: referenceOf(module, name, digest)}) + } + return &b +} + +// FirstViews is the node-engine's first views as the seat's verb first-views lists them (§698): the views it follows (none: it +// follows no branch), the entries it keeps and the wants it composes from. +func (c *Composer) FirstViews() (views []map[string]any, entries []EntryRow, wants []WantRow) { + views, entries, wants = []map[string]any{}, []EntryRow{}, []WantRow{} + if !c.byEntries() { + return + } + seen := map[string]bool{} + for _, r := range c.entries.Rows() { + seen[r.Fingerprint] = true + entries = append(entries, EntryRow{Fingerprint: r.Fingerprint, State: string(r.State), Since: r.Since}) + } + c.mu.Lock() + for fp, k := range c.keptEnt { + if !seen[fp] { + entries = append(entries, EntryRow{Fingerprint: fp, State: string(meshcatalogclient.EntryAnswered), Since: k.since}) + } + } + for module, w := range c.wants { + if module == agent.NodeModule || w.Fp == "" { + continue + } + wants = append(wants, WantRow{Module: module, Fingerprint: w.Fp, KeptSince: c.keptWantVal[module].Since}) + } + c.mu.Unlock() + sort.Slice(entries, func(i, j int) bool { return entries[i].Fingerprint < entries[j].Fingerprint }) + sort.Slice(wants, func(i, j int) bool { return wants[i].Module < wants[j].Module }) + return +} + +// EntryAgain asks fp's entry of the catalog half again now (the seat's verb entry-again, §698): one the composer keeps, whatever +// its state. The entry kept of it is dropped, so a pass holds the module until the answer. +func (c *Composer) EntryAgain(ctx context.Context, fp string) error { + if !c.byEntries() { + return fmt.Errorf("this engine reads no entry by fingerprint") + } + c.mu.Lock() + _, kept := c.keptEnt[fp] + delete(c.keptEnt, fp) + c.mu.Unlock() + held := kept + for _, r := range c.entries.Rows() { + held = held || r.Fingerprint == fp + } + if !held { + return fmt.Errorf("no entry of %s is kept", fp) + } + if !kept { + return c.entries.Again(ctx, fp) + } + c.entries.Need(ctx, fp) // makes the row and asks once + c.askRehand() + return nil +} + +// EntryCall is env.Entry over the engine's link session: the seat mesh-catalog's verb `entry`, read in the engine's own name +// (the manifest's invokes names it). A refusal in the verb's own words that it keeps no entry of the build is the holder's +// answer that it holds none, and is told so; any other failure is a call that did not reach the holder, and is told so too. +func EntryCall(ask func(ctx context.Context, seat, verb string, args any) (json.RawMessage, error)) meshcatalogclient.Call { + return func(key string, body any) (json.RawMessage, error) { + seatName, verb, _ := strings.Cut(strings.TrimPrefix(key, "seat:"), ".") + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + raw, err := ask(ctx, seatName, verb, body) + if err != nil && strings.Contains(err.Error(), noEntryWords) { + out, _ := json.Marshal(map[string]string{"error": err.Error()}) + return out, nil + } + return raw, err + } +} + +// noEntryWords are the words the seat's holder refuses an unknown fingerprint in. +const noEntryWords = "no entry of fingerprint" diff --git a/internal/selfcompose/entries_test.go b/internal/selfcompose/entries_test.go new file mode 100644 index 00000000..7136b509 --- /dev/null +++ b/internal/selfcompose/entries_test.go @@ -0,0 +1,145 @@ +package selfcompose + +import ( + "context" + "encoding/json" + "errors" + "strings" + "sync" + "testing" + + "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/contracts/go/clients/meshcatalogclient" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshcatalog" + "git.novox.be/novox/mesh-delivery/contracts/go/seats/meshwants" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/builds" + "github.com/novox/mesh-host/internal/agent" + "github.com/novox/mesh-host/internal/buckets" + "github.com/novox/mesh-host/internal/buckets/bucketstest" + "github.com/novox/mesh-host/internal/compositions" +) + +// Design 47 G §421, §683, §698 (the lead's answers A-47, A-50): the node-engine reads the entry its want names by the +// fingerprint, through its own calls of the seat mesh-catalog's verb entry, and reads neither the indexer's states nor the +// builder's; a want whose entry the catalog half holds none of is held, its have saying entry-missing and naming the +// fingerprint; the entries and wants it composes from are kept in its composition state, written before the composition that +// uses them and read at its start before anything is asked; a want removed takes its kept want and entry away. +func composeFromEntries(t *testing.T) { + b := newFakeBus() + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) + ghost := meshcatalogclient.SourcePrefix2 + strings.Repeat("b", 64) + var manifest map[string]any + if err := json.Unmarshal([]byte(lemurs), &manifest); err != nil { + t.Fatal(err) + } + entry := meshcatalog.CatalogEntry{Module: "lemurs", Fp: fp, Manifest: manifest, Repository: "novox/x", Commit: "c", + ManifestDigest: meshcatalogclient.ManifestDigest([]byte(lemurs))} + var mu sync.Mutex + var asked []string + ask := func(_ context.Context, seat, verb string, args any) (json.RawMessage, error) { + in := args.(meshcatalog.EntryInput) + mu.Lock() + asked = append(asked, in.Fingerprint) + mu.Unlock() + if seat != "mesh-catalog" || verb != "entry" { + t.Errorf("asked %s.%s", seat, verb) + } + if in.Fingerprint != fp { + return nil, errors.New("no entry of fingerprint " + in.Fingerprint + " is kept") + } + return json.Marshal(meshcatalog.EntryAnswer{Entry: entry}) + } + state := &fakeComposed{kv: map[string]ComposedEntry{}} + var held map[string]string + env := b.env() + env.Entry, env.ComposedState = EntryCall(ask), state + env.Refused = func(_ string, h map[string]string) { mu.Lock(); held = h; mu.Unlock() } + c := New(env) + out, err := c.Watch(ctx, "n1.>") + if err != nil { + t.Fatal(err) + } + place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, + Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}, {Module: "ghost"}}}) + have, _ := nodeengineclient.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) + nodeWant, _ := bucketstest.NodeWant(meshwants.WantsByWantNodeNodeValue{Self: true}) + b.putAssigned(t, place) + b.put(t, buckets.Have, "n1.node", have) + b.put(t, buckets.Wants, "want.n1.node", nodeWant) + for module, w := range map[string]meshwants.WantsByWantNodeModuleValue{"lemurs": {Fp: fp, Self: true}, "ghost": {Fp: ghost, Self: true}} { + raw, _ := bucketstest.Want(w) + b.put(t, buckets.Wants, "want.n1."+module, raw) + } + for _, bucket := range []string{buckets.Wants, buckets.Assignments, buckets.Have, ResolvedBucket} { + b.send(bucket, agent.Update{Caught: true}) + } + lemursKey, _ := compositions.Key("n1", "lemurs", fp) + handed := map[string]bool{} + for !handed[lemursKey] { + handed[next(t, out).Key] = true + } + // 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 { + t.Errorf("the node-engine read %s of %s", r[1], r[0]) + } + } + b.mu.Unlock() + // A want whose entry the half holds none of is held, saying entry-missing and the fingerprint; the rest is composed. + mu.Lock() + why := held["ghost"] + mu.Unlock() + if !strings.HasPrefix(why, agent.EntryMissing) || !strings.Contains(why, ghost) || held["lemurs"] != "" { + t.Errorf("held %q, want ghost held with %s naming %s", held, agent.EntryMissing, ghost) + } + // Each entry and want it composes from is kept in the composition state before the composition that uses them. + state.mu.Lock() + _, hasEntry := state.kv[entryKey(fp)] + _, hasWant := state.kv["want.lemurs"] + state.mu.Unlock() + if !hasEntry || !hasWant { + t.Errorf("the composition state holds entry %v and want %v of lemurs: %v", hasEntry, hasWant, state.written()) + } + // The seat's first-views lists them. + _, entries, wants := c.FirstViews() + if len(entries) < 2 || len(wants) != 2 { + t.Errorf("first-views lists %v and %v", entries, wants) + } + // A restart reads what is kept before it asks anything: lemurs's entry is not asked again. + mu.Lock() + asked = nil + mu.Unlock() + env2 := b.env() + env2.Entry, env2.ComposedState = EntryCall(ask), state + c2 := New(env2) + if _, err := c2.Watch(ctx, "n1.>"); err != nil { + t.Fatal(err) + } + c2.mu.Lock() + kept, keptWant := c2.keptEnt[fp], c2.wants["lemurs"].Fp + c2.mu.Unlock() + if kept.entry.Module != "lemurs" || keptWant != fp { + t.Errorf("a restart read entry %q and want %q from the state", kept.entry.Module, keptWant) + } + // A want removed takes its kept want and entry away. + raw, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Removed: true}) + b.put(t, buckets.Wants, "want.n1.lemurs", raw) + for range 20 { + state.mu.Lock() + _, hasWant = state.kv["want.lemurs"] + _, hasEntry = state.kv[entryKey(fp)] + state.mu.Unlock() + if !hasWant && !hasEntry { + break + } + next(t, out) + } + if hasWant || hasEntry { + t.Errorf("a removed want left its kept want %v and entry %v", hasWant, hasEntry) + } +} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index d4c84594..91753a13 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -126,6 +126,10 @@ type Env struct { // SeatListChanged says the reader's held set changed: the node composes again, so what waited for the list is // handed at the read and not on a timer. Nil hears nothing. SeatListChanged <-chan struct{} + // Entry asks the seat mesh-catalog's verb `entry` of a fingerprint (meshcatalogclient.EntryVerb), in the engine's own name. + // Set, the composer reads the catalog half by the fingerprints its wants and the runnings it heard name and nothing else of + // the indexer's or the builder's states (design 47 G §421, §628); nil reads the indexer's states, as the shadow does. + Entry meshcatalogclient.Call } // ManifestOf is the digest of the manifest of the build a module runs at fp, from the composer's picture of the catalog half; @@ -175,6 +179,14 @@ type Composer struct { notTaken map[string]string // unmadeSaid is, by module, the stable part of the reason its last journal line said (without where the picture stands). unmadeSaid map[string]string + // entries is the cache of entries asked by fingerprint, nil where the composer reads the indexer's states; keptEnt the entries + // held from the composition state or an answer, keptWantVal the wants it keeps there and keptWants the modules whose want + // stands from that state alone until the live read of the wants answers (entries.go). Under mu; keptWrite orders the writes. + entries *meshcatalogclient.Entries + keptEnt map[string]keptEntry + keptWantVal map[string]keptWant + keptWants map[string]bool + keptWrite sync.Mutex // 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 @@ -374,6 +386,12 @@ func New(env Env) *Composer { if env.Say == nil { env.Say = func(string) {} } + c := newComposer(env) + c.newEntries() + return c +} + +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{}, @@ -437,6 +455,21 @@ type input struct { } 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}, @@ -491,6 +524,8 @@ func (c *Composer) takeBuildDone(u agent.Update) bool { // input's watch ends. func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, error) { ins := c.inputs() + // What the node kept of the entries and the wants it composes from is read before anything is asked (§683). + c.loadKept(ctx) type tagged struct { in int u agent.Update @@ -598,6 +633,9 @@ func (c *Composer) Watch(ctx context.Context, _ string) (<-chan agent.Update, er dirty = true // a secret changed: its new generation is asked of the vault at the compose case t.u.Caught: caught[t.in] = true + if ins[t.in].bucket == buckets.Wants { + c.endKeptWants() // the wants the seat says now are the ones composed from (§683) + } case ins[t.in].changed != nil: dirty = ins[t.in].changed(c, t.u) || dirty default: @@ -759,6 +797,7 @@ func (c *Composer) takeWantLocked(u agent.Update) string { return "" } delete(c.notTaken, module) + delete(c.keptWants, module) w, err := meshwants.DecodeWantsByWantNodeModuleValue(u.Value) if u.Deleted || err != nil || w.Removed || w.Fp == "" { delete(c.wants, module) @@ -960,6 +999,7 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { } } }() + c.syncEntries(ctx) c.mu.Lock() fps := map[string]string{} kept := map[string]string{} @@ -1024,6 +1064,15 @@ func (c *Composer) compose(ctx context.Context) []agent.Update { // (ADR 0636 §1): the module is held as it runs before anything of its build is read, so no take-in or artifact // reference of it (the artifact store's seat is not bootstrapped) can refuse the pass (issue 1966). scope.Held = c.heldUntilListed(ctx, r, own) + // A want whose entry the catalog half has not answered, or holds none of, is held with what runs of it (§421). + if holds := c.entryHolds(); len(holds) > 0 { + if scope.Held == nil { + scope.Held = map[string]string{} + } + for module, why := range holds { + scope.Held[module] = why + } + } missed := c.missed.Load() compositions, skipped, held, err := frombus.Compose(ctx, r, c.env.Mesh, scope, fps) err = c.missedIn(missed, err) @@ -1533,6 +1582,9 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { if ok { return raw, nil } + if r.c.byEntries() { + return nil, fmt.Errorf("the catalog half's entries hold no manifest %s", digest) + } key, err := meshcatalogclient.ManifestsKey(digest) if err != nil { return nil, err @@ -1554,6 +1606,9 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { } func (r reader) Build(ctx context.Context, module, fp string) (*builds.BuilderBuild, error) { + if r.c.byEntries() { + return r.c.buildOf(module, fp), nil + } key, err := builds.BuilderBuildKey(module, fp) if err != nil { return nil, err diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index f6d1f386..af655ddb 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -138,7 +138,7 @@ func next(t *testing.T, out <-chan agent.Update) agent.Update { case u := <-out: return u case <-time.After(5 * time.Second): - t.Fatal("no update from the composer") + t.Fatalf("no update from the composer") return agent.Update{} } } @@ -156,6 +156,7 @@ func quiet(t *testing.T, out <-chan agent.Update) { // every input was read, hands each, then Caught; an input that changes nothing composed hands nothing; a module no want // names any more is taken back; a node that cannot compose says so once and hands nothing. func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { + t.Run("it reads the catalog half by the fingerprints its wants name", composeFromEntries) b := newFakeBus() fp := meshcatalogclient.SourcePrefix2 + strings.Repeat("a", 64) digest := meshcatalogclient.ManifestDigest([]byte(lemurs)) diff --git a/module.json b/module.json index 82b00dec..3d95bb27 100644 --- a/module.json +++ b/module.json @@ -21,6 +21,7 @@ } ], "invokes": [ + "seat:mesh-catalog.entry", "seat:mesh-catalog.seat-definitions" ], "consumes": [