From e558510a54ac28b0d4476bb4ac9d96132ce643d3 Mon Sep 17 00:00:00 2001 From: jochen Date: Thu, 12 Nov 2026 22:28:15 +0100 Subject: [PATCH] H5 (review r2, A-308, A-312): the shadow watches no indexer modules; it reads the entries its wants name and asks the verb module for a module a candidate adds MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Builds: design 47 G §421, §628 --- internal/selfcompose/candidate.go | 1 + internal/selfcompose/candidate_test.go | 16 +++- internal/selfcompose/entries.go | 76 +++++++++++++++++- internal/selfcompose/selfcompose.go | 9 ++- internal/selfcompose/selfcompose_test.go | 53 +++++++++---- internal/selfcompose/shadow.go | 99 +++++++++++++++++++++--- module.json | 1 + packaging/operator_keys_test.go | 2 +- 8 files changed, 226 insertions(+), 31 deletions(-) diff --git a/internal/selfcompose/candidate.go b/internal/selfcompose/candidate.go index 502756d9..53161098 100644 --- a/internal/selfcompose/candidate.go +++ b/internal/selfcompose/candidate.go @@ -328,6 +328,7 @@ func (c *Composer) scratch() *Composer { out.certified = c.certified out.assignmentKeys = c.assignmentKeys out.origin = c + out.forCandidates = c.forCandidates return out } diff --git a/internal/selfcompose/candidate_test.go b/internal/selfcompose/candidate_test.go index e78885aa..dcebd22f 100644 --- a/internal/selfcompose/candidate_test.go +++ b/internal/selfcompose/candidate_test.go @@ -7,6 +7,7 @@ import ( "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" + "git.novox.be/novox/mesh-delivery/modules/mesh-delivery-contracts/go/seats/meshwants" "github.com/novox/mesh-host/internal/buckets" "github.com/novox/mesh-host/internal/buckets/bucketstest" "strings" @@ -78,6 +79,15 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) { env.ComposedState = state var applied atomic.Int64 env.Refused = func(string, map[string]string) { applied.Add(1) } + // What the node's want names is read from the catalog half's entry (design 47 G §421, A-308): the shadow watches no modules. + 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: digest} + env.Entry = EntryCall(func(_ context.Context, _, _ string, _ any) (json.RawMessage, error) { + return json.Marshal(meshcatalog.EntryAnswer{Entry: entry}) + }) var handed atomic.Int64 s := NewShadow(env, func(context.Context, map[string]nodeengine.CandidateAnswer) { handed.Add(1) }) ctx, cancel := context.WithCancel(context.Background()) @@ -89,7 +99,7 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) { place := assignedValues(t, meshmoduleassigner.Assignments{Node: "n1", Decided: meshmoduleassigner.NodeDecided{Hub: true, Address: "10.77.0.1"}, Modules: []meshmoduleassigner.PlacedModule{{Module: "lemurs"}}}) have, _ := nec.EncodeHaveNode(nodeengine.HaveByNodeNodeValue{Record: &nodeengine.NodeRecord{Node: "n1", TunnelKey: "k", SealingKey: "sk", Capabilities: []string{"systemd"}}}) - mod, _ := meshcatalogclient.EncodeModule(meshcatalog.ModulesByModuleValue{Fp2: fp, Manifest: digest, Repository: "novox/x", Commit: "c"}) + want, _ := bucketstest.Want(meshwants.WantsByWantNodeModuleValue{Fp: fp, Self: true}) deadline := time.Now().Add(5 * time.Second) watching := func() int { b.mu.Lock(); defer b.mu.Unlock(); return len(b.watches) } for watching() < 6 && time.Now().Before(deadline) { @@ -97,8 +107,8 @@ func TestTheNodeComposesACandidateOnAsk(t *testing.T) { } b.putAssigned(t, place) b.put(t, buckets.Have, "n1.node", have) - b.put(t, buckets.Modules, "lemurs", mod) - for _, bucket := range []string{buckets.ProposalCandidates, buckets.Assignments, buckets.Have, buckets.Modules, ResolvedBucket} { + b.put(t, buckets.Wants, "want.n1.lemurs", want) + for _, bucket := range []string{buckets.ProposalCandidates, buckets.Assignments, buckets.Have, buckets.Wants, ResolvedBucket} { b.send(bucket, agent.Update{Caught: true}) } for !s.ready.Load() && time.Now().Before(deadline) { diff --git a/internal/selfcompose/entries.go b/internal/selfcompose/entries.go index 979175de..8c427341 100644 --- a/internal/selfcompose/entries.go +++ b/internal/selfcompose/entries.go @@ -6,6 +6,7 @@ import ( "encoding/hex" "encoding/json" "fmt" + "git.novox.be/novox/mesh-sdk/go/buscalls" "reflect" "sort" "strings" @@ -257,7 +258,10 @@ func (c *Composer) syncEntries(ctx context.Context) { c.entryOf(fp) } for _, row := range c.entries.Rows() { - if _, named := need[row.Fingerprint]; !named { + c.mu.Lock() + added := c.addedFps[row.Fingerprint] + c.mu.Unlock() + if _, named := need[row.Fingerprint]; !named && !added { c.entries.Release(row.Fingerprint) c.tellNotFound(row.Fingerprint, false) } @@ -265,7 +269,7 @@ func (c *Composer) syncEntries(ctx context.Context) { c.mu.Lock() var drop []string for fp := range c.keptEnt { - if _, named := need[fp]; !named { + if _, named := need[fp]; !named && !c.addedFps[fp] { delete(c.keptEnt, fp) drop = append(drop, entryKey(fp)) } @@ -568,3 +572,71 @@ func (c *Composer) FormChanges(_ context.Context, module, runningFp, appliedFp s } return catalogue.FormChanges(running, applied), nil } + +// syncShadowEntries asks the entry of each build a want of this node names, drops the one nothing names, lays c.modules and +// c.manifests from the entries held, and says whether an entry asked is not yet answered: the shadow's picture of the catalog, +// where it once watched the indexer's modules (design 47 G §421, A-308). +func (c *Composer) syncShadowEntries(ctx context.Context) (pending bool) { + if !c.byEntries() { + return false + } + c.syncEntries(ctx) + c.mu.Lock() + defer c.mu.Unlock() + for module, w := range c.wants { + if module == agent.NodeModule || !strings.HasPrefix(w.Fp, meshcatalogclient.SourcePrefix2) { + continue + } + if _, held := c.keptEnt[w.Fp]; !held { + for _, r := range c.entries.Rows() { + if r.Fingerprint == w.Fp && r.State == meshcatalogclient.EntryAsked { + return true + } + } + } + } + return false +} + +// moduleVerb is the seat mesh-catalog's verb `module`: one module's trunk fingerprint and manifest by name. +const moduleVerb = "seat:" + meshcatalog.Seat + "." + meshcatalog.VerbModule + +// addedModule is the trunk's build of a module an assignment candidate adds to this node when no want here names it yet: the +// fingerprint the seat's verb `module` answers, then its entry by the verb `entry` as for a wanted module (design 47 G §421, A-312). +// A call, not a watch: nothing of the catalog is kept. It answers the entry when held; otherwise why not: a module with no +// fingerprint says the verb's reason, an entry asked says so (its answer wakes the shadow, which composes again). +func (c *Composer) addedModule(ctx context.Context, module string) (fp string, entry meshcatalog.CatalogEntry, held, pending bool, why string) { + if c.env.Entry == nil { + return "", entry, false, false, "this node reads no module of the catalog half" + } + raw, err := c.env.Entry(moduleVerb, meshcatalog.ModuleInput{Module: module}) + if err != nil { + return "", entry, false, false, fmt.Sprintf("the catalog half's answer for %s: %v", module, err) + } + answer, said, ok := buscalls.Unwrap(raw) + if !ok { + return "", entry, false, false, fmt.Sprintf("the catalog half holds no module %s: %s", module, said) + } + var a meshcatalog.ModuleAnswer + if err := json.Unmarshal(answer, &a); err != nil { + return "", entry, false, false, fmt.Sprintf("the catalog half's answer for %s cannot be read: %v", module, err) + } + if a.Module.Fp == "" { + return "", entry, false, false, fmt.Sprintf("the catalog half gives %s no fingerprint: %s", module, a.Module.None) + } + fp = a.Module.Fp + 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 state { + case meshcatalogclient.EntryAnswered: + return fp, e, true, false, "" + case meshcatalogclient.EntryNotFound: + return fp, e, false, false, fmt.Sprintf("the catalog half holds no entry of %s's build %s", module, fp) + } + 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) +} diff --git a/internal/selfcompose/selfcompose.go b/internal/selfcompose/selfcompose.go index ea0d7af1..07999e2c 100644 --- a/internal/selfcompose/selfcompose.go +++ b/internal/selfcompose/selfcompose.go @@ -198,6 +198,11 @@ type Composer struct { keptWantVal map[string]keptWant keptWants map[string]bool keptWrite sync.Mutex + // forCandidates marks the shadow's composers: they read the catalog half's entries for the modules their wants name (§421) and, for + // a candidate's own builds, the manifests and verdicts of the indexer that compose-candidate alone keeps (§628, A-308). + 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 @@ -1631,7 +1636,7 @@ func (r reader) Manifest(ctx context.Context, digest string) ([]byte, error) { if ok { return raw, nil } - if r.c.byEntries() { + if r.c.byEntries() && !r.c.forCandidates { return nil, fmt.Errorf("the catalog half's entries hold no manifest %s", digest) } key, err := meshcatalogclient.ManifestsKey(digest) @@ -1655,7 +1660,7 @@ 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() { + if r.c.byEntries() && !r.c.forCandidates { return r.c.buildOf(module, fp), nil } key, err := builds.BuilderBuildKey(module, fp) diff --git a/internal/selfcompose/selfcompose_test.go b/internal/selfcompose/selfcompose_test.go index cf9eff83..4face352 100644 --- a/internal/selfcompose/selfcompose_test.go +++ b/internal/selfcompose/selfcompose_test.go @@ -1127,6 +1127,34 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { "keeper": {Fp2: fp, Manifest: kdigest, Repository: "novox/k", Commit: "c"}} readOnly := &fakeBus{kv: map[string][]byte{buckets.Manifests + "/" + key: man, buckets.Manifests + "/" + kkey: kman}} senv := readOnly.env() + // The catalog half: lemurs's entry, which its want names, and keeper's, which a candidate adding it asks by the verb module + // (design 47 G §421, A-308, A-312). The shadow watches no modules. + kfp := meshcatalogclient.SourcePrefix2 + strings.Repeat("c", 64) + entryOfManifest := func(module, fp, body string) meshcatalog.CatalogEntry { + var m map[string]any + if err := json.Unmarshal([]byte(body), &m); err != nil { + t.Fatal(err) + } + 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)} + senv.Entry = EntryCall(func(_ context.Context, _, verb string, args any) (json.RawMessage, error) { + switch in := args.(type) { + case meshcatalog.ModuleInput: + 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}}) + } + } + return nil, fmt.Errorf("the catalog holds no module %s", in.Module) + case meshcatalog.EntryInput: + if e, ok := catalogEntries[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) + }) senv.Watch = sb.watch var ownFails atomic.Bool // the vault's own on a bus that does not answer (2026-10-28, ace's answer to seed-7) senv.Own = func(context.Context, map[string]any) (json.RawMessage, error) { @@ -1177,10 +1205,6 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { tp := assignedValues(t, trunk) sb.putAssigned(tp, 1) sb.put(buckets.Have, "n1.node", rec, 1) - for name, m := range mods { - raw, _ := meshcatalogclient.EncodeModule(m) - sb.put(buckets.Modules, name, raw, 1) - } wantLemurs := meshwants.WantsByWantNodeModuleKey("n1", "lemurs") var lemursDigest string for _, c := range running { @@ -1244,12 +1268,18 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { } } waitFor("every group answered", func(a map[string]nodeengine.CandidateAnswer) bool { return len(a) == 6 }) + // keeper, which the candidate adds and no want names, is asked of the catalog half (its entry by the verb entry after the verb + // module): the answer waits for it, and is the same as before the shadow watched no modules once it comes (A-312). + waitFor("the added module's entry answered", func(a map[string]nodeengine.CandidateAnswer) bool { return len(a["adds"].Added) == 1 }) // Every watch and get the shadow makes is one the node's grant allows (issue 1126, 2026-10-27 09:21). sb.mu.Lock() for _, w := range sb.watches { if !grantAllows(t, "n1", w.bucket, w.pattern) { t.Errorf("the shadow watches %s of %s, which the node's grant does not allow", w.pattern, w.bucket) } + if w.bucket == buckets.Modules { + t.Errorf("the shadow watches the indexer's modules (%s), which design 47 G A-308 ends", w.pattern) + } } sb.mu.Unlock() readOnly.mu.Lock() @@ -1334,8 +1364,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { lemursHave("its tool lemurs_health: no answer within 10s", 11) lemursHave("its tool lemurs_health: no answer within 20s", 12) sb.putAssigned(tp, 2) - lraw, _ := meshcatalogclient.EncodeModule(mods["lemurs"]) - sb.put(buckets.Modules, "lemurs", lraw, 2) + sb.put(buckets.Wants, wantLemurs, wantRaw, 2) drain() if got := answering(); got != settled { t.Errorf("answered %d more time(s) with no candidate revision and no input of the node changed", got-settled) @@ -1393,7 +1422,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { // Review of L2b, round 2: a watch ends while the link stays; a group's key is deleted before the shadow watches // again. The new run reads afresh, so that group is no longer answered. sb.missing(buckets.ProposalCandidates, true) - sb.end(buckets.Modules) + sb.end(buckets.Wants) dropsKey := meshmoduleassigner.CandidatesByProposalNodeKey("drops", "n1") sb.del(buckets.ProposalCandidates, dropsKey) time.Sleep(30 * time.Millisecond) @@ -1428,6 +1457,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { nenv := readOnly.env() nenv.Watch = nb.watch nenv.Running = senv.Running + nenv.Entry = senv.Entry var nsaid []string var nmu sync.Mutex nenv.Say = func(l string) { nmu.Lock(); nsaid = append(nsaid, l); nmu.Unlock() } @@ -1439,10 +1469,7 @@ func TestTheNodeComposesItsWantsFromTheBus(t *testing.T) { nb.made(6) nb.putAssigned(tp, 1) nb.put(buckets.Have, "n1.node", rec, 1) - for name, m := range mods { - raw, _ := meshcatalogclient.EncodeModule(m) - nb.put(buckets.Modules, name, raw, 1) - } + nb.put(buckets.Wants, wantLemurs, wantRaw, 1) sameRaw, _ := bucketstest.Candidate(meshmoduleassigner.CandidatesByProposalNodeValue{Proposal: "same", Assignments: &trunk}) sameKey := meshmoduleassigner.CandidatesByProposalNodeKey("same", "n1") nb.put(buckets.ProposalCandidates, sameKey, sameRaw, 3) @@ -1450,7 +1477,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.Modules, builds.BuilderBuildsBucket} { + for _, bucket := range []string{buckets.ProposalCandidates, buckets.Assignments, buckets.Have, buckets.Wants, builds.BuilderBuildsBucket} { nb.send(bucket, agent.Update{Caught: true}) } select { @@ -1825,7 +1852,7 @@ func TestAHaveSaysWhetherItsModuleRunsInsideTheNodesOwn(t *testing.T) { } s := &Shadow{c: c, ran: map[string]string{}} s.takeRan(agent.Update{Key: key, Value: value}) - if _, got := s.heldToday()["web"]; got != row.heldToday { + if _, got := s.heldToday(nil)["web"]; got != row.heldToday { t.Errorf("web held today: %v, want %v", got, row.heldToday) } }) diff --git a/internal/selfcompose/shadow.go b/internal/selfcompose/shadow.go index eaf7ab1e..e272cf8c 100644 --- a/internal/selfcompose/shadow.go +++ b/internal/selfcompose/shadow.go @@ -8,12 +8,14 @@ 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-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" "github.com/novox/mesh-host/internal/buckets" "github.com/novox/mesh-host/internal/compose/frombus" engcomp "github.com/novox/mesh-host/internal/compositions" "github.com/novox/mesh-host/internal/judge" + "maps" "sort" "strings" "sync" @@ -61,6 +63,8 @@ type Shadow struct { // after a bound (laterAgain), whether or not an input moves (issue 1126, 2026-10-28: ace kept "the vault's own: not // yet ... no responders" in its answer to seed-7). owed bool + // entryMoved wakes the run when an entry it asked of the catalog half is answered or found missing (§421). + entryMoved chan struct{} // ready says this run read every input once: the verb compose-candidate composes only then (ComposeCandidate). ready atomic.Bool } @@ -72,7 +76,7 @@ type answeredAt struct { } // inputWords name the shadow's inputs, in Run's order, as an answer says which of them woke it. -var inputWords = [...]string{"the candidate", "assignment", "records", "modules", "values", "haves"} +var inputWords = [...]string{"the candidate", "assignment", "records", "wants", "values", "haves"} // wokenBy says what moved between the answer a proposal stands at (had, when done) and the one about to be composed (at): // its first answer, a retry, its candidate's revisions, or the inputs that changed, by their words. @@ -106,8 +110,28 @@ type candidateKey struct { // NewShadow is the shadow composer of one node; answer writes the node's answers into its own have key. func NewShadow(env Env, answer func(ctx context.Context, answers map[string]nodeengine.CandidateAnswer)) *Shadow { - return &Shadow{c: New(env), answer: answer, candidates: map[string]map[string]candidateKey{}, - answers: map[string]nodeengine.CandidateAnswer{}, ran: map[string]string{}} + s := &Shadow{answer: answer, candidates: map[string]map[string]candidateKey{}, + answers: map[string]nodeengine.CandidateAnswer{}, ran: map[string]string{}, entryMoved: make(chan struct{}, 1)} + s.c = s.composer(env) + return s +} + +// composer is a composer for the shadow: it asks the catalog half's entry of each build the node's wants name, and wakes the +// run when one answers (design 47 G §421, A-308: the node-engine no longer watches the indexer's modules). +func (s *Shadow) composer(env Env) *Composer { + env.ComposedState = nil // the node's own composer keeps the composition state; the shadow writes nothing + c := New(env) + c.forCandidates = true + if c.entries != nil { + c.entries.Changed = func(row meshcatalogclient.EntryRow) { + c.entryChanged(row) + select { + case s.entryMoved <- struct{}{}: + default: + } + } + } + return c } // Keep runs the shadow until ctx ends, beginning again after backoff(try) whenever it stops: an input it cannot watch @@ -153,7 +177,7 @@ func (s *Shadow) Run(ctx context.Context) error { s.answered = map[string]answeredAt{} s.moved = [len(inputWords)]uint64{} s.retry = map[string]bool{} - s.c = New(s.c.env) + s.c = s.composer(s.c.env) s.mu.Unlock() s.ready.Store(false) defer s.ready.Store(false) @@ -164,7 +188,7 @@ func (s *Shadow) Run(ctx context.Context) error { {buckets.ProposalCandidates, ">", nil, nil}, {buckets.Assignments, ">", (*Composer).takeAssignments, nil}, {buckets.Have, "*." + nodeengineclient.NodeWord, nil, nil}, - {buckets.Modules, ">", (*Composer).takeModule, nil}, + {buckets.Wants, "want." + node + ".>", (*Composer).takeWant, nil}, {ResolvedBucket, node + ".>", (*Composer).takeValue, nil}, // What this node runs moved: an apply ends in its module's have. {buckets.Have, node + ".>", nil, nil}, @@ -209,6 +233,9 @@ func (s *Shadow) Run(ctx context.Context) error { case <-again: again = nil dirty = true + case <-s.entryMoved: + s.moved[3]++ + dirty = true case <-waiting.C: if !ready { s.notReady(ctx, ins, caught) @@ -245,10 +272,15 @@ func (s *Shadow) Run(ctx context.Context) error { continue } } + if !ready && !all(caught) { + continue + } + // What the wants name is read from the catalog half's entries; an answer waits until each entry asked has come (§421). + if (dirty || !ready) && s.c.syncShadowEntries(ctx) { + continue + } if !ready { - if ready = all(caught); !ready { - continue - } + ready = true s.ready.Store(true) dirty = true } @@ -386,6 +418,12 @@ func (s *Shadow) answerAll(ctx context.Context) { proposals = append(proposals, l) } // A proposal whose keys' revisions and the node's inputs are those its answer stands at is not answered again. + if len(s.candidates) == 0 { + // No proposal is left that adds a module: the entries asked for those are no longer kept (A-312). + c.mu.Lock() + c.addedFps = nil + c.mu.Unlock() + } for g := range s.answered { if _, ok := s.candidates[g]; !ok { delete(s.answered, g) @@ -554,6 +592,46 @@ func (s *Shadow) answerProposal(ctx context.Context, proposal string) (nodeengin inOwn := c.inOwnOf(*assignment) c.mu.Unlock() fps := shadowBuilds(running, *assignment, modules, inOwn) + // 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{} + folded := map[string]bool{} + for _, m := range inOwn { + folded[m] = true + } + for _, pm := range assignment.Modules { + if _, has := fps[pm.Module]; has || folded[pm.Module] || pm.Module == "" { + continue + } + fp, entry, held, pending, why := c.addedModule(ctx, pm.Module) + if pending { + // Its answer wakes the shadow (entryMoved), which composes this proposal again: until then this is no answer to keep, as one + // that met a read the bus did not answer is none. + c.missed.Add(1) + return nodeengine.CandidateAnswer{Proposal: proposal, Revision: int64(rev), Verdict: nodeengine.CandidateAnswerVerdictRefused, + Refused: why + ", so this answer waits for it"}, true + } + if !held { + heldAdded[pm.Module] = why + continue + } + digest, raw, err := entryManifest(entry) + if err != nil { + 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 + } + 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 { + m.Problems, m.Warnings, m.Unknown = entry.Inputs.Problems, entry.Inputs.Warnings, entry.Inputs.Unknown + } + c.mu.Lock() + c.modules[pm.Module] = m + c.manifests[digest] = raw + c.mu.Unlock() + modules[pm.Module] = m + fps[pm.Module] = fp + } kept := map[string]string{} for module, fp := range fps { if mod, ok := modules[module]; ok && mod.Fp2 == fp { @@ -571,7 +649,7 @@ func (s *Shadow) answerProposal(ctx context.Context, proposal string) (nodeengin c.wants = held c.mu.Unlock() 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()}, fps, running, proposal, rev) + SameValue: c.sameValue, HeldToday: s.heldToday(heldAdded)}, fps, running, proposal, rev) s.mu.Lock() ran := make(map[string]string, len(s.ran)) for m, d := range s.ran { @@ -583,8 +661,9 @@ func (s *Shadow) answerProposal(ctx context.Context, proposal string) (nodeengin // heldToday is what the node holds today: each module whose have's phase is held, and what its own composer holds, the // node's own composition ("node") among them when it composes none of its own. -func (s *Shadow) heldToday() map[string]string { +func (s *Shadow) heldToday(added map[string]string) map[string]string { out := map[string]string{} + maps.Copy(out, added) s.mu.Lock() for m, how := range s.heldNow { if how != "" { diff --git a/module.json b/module.json index a2d7d50d..e749c563 100644 --- a/module.json +++ b/module.json @@ -41,6 +41,7 @@ ], "invokes": [ "seat:mesh-catalog.entry", + "seat:mesh-catalog.module", {"verb": "seat:mesh-catalog.revision", "fields": ["none-yet"]}, "seat:mesh-catalog.seat-definitions", "seat:mesh-grants.withheld", diff --git a/packaging/operator_keys_test.go b/packaging/operator_keys_test.go index 6d07d6f6..12524586 100644 --- a/packaging/operator_keys_test.go +++ b/packaging/operator_keys_test.go @@ -78,7 +78,7 @@ func TestTheManifestConsumesTheBackupEvent(t *testing.T) { t.Errorf("uses is %v, want none: the invoke of seat:mesh-grants.withheld is the edge (§635, A-307)", m.Uses) } for _, v := range []any{"seat:mesh-recovery.kept-points", "seat:mesh-recovery.restore-kept", "seat:mesh-recovery.snapshot-kept", - "seat:mesh-grants.withheld", map[string]any{"verb": "seat:mesh-catalog.revision", "fields": []any{"none-yet"}}} { + "seat:mesh-grants.withheld", "seat:mesh-catalog.module", map[string]any{"verb": "seat:mesh-catalog.revision", "fields": []any{"none-yet"}}} { if !slices.ContainsFunc(m.Invokes, func(have any) bool { return reflect.DeepEqual(have, v) }) { t.Errorf("invokes %v lacks %v", m.Invokes, v) }