diff --git a/.gitignore b/.gitignore index 48b8bf9..420d8ac 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,4 @@ vendor/ +# A compiled program never enters a commit (issue 1577): `go build .` in a command directory leaves one there. +modules/*/cmd/*/mesh-* +!modules/*/cmd/*/mesh-*.go diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go index eac7f57..d6d3157 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go @@ -172,6 +172,9 @@ func TestTheAssignerStartsAndSaysWhatItKeeps(t *testing.T) { // 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket. func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { for _, missing := range readBuckets { + if missing == delivery.PlacementBucket { + continue // it starts without it, below + } b := startBed(t, missing) select { case err := <-b.start(t): @@ -218,6 +221,34 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { t.Fatal("the put was not taken further once its bucket was made") } } + + // Review of L2 (D1): without the indexer's placement state it starts all the same, and once the bucket is made (the + // indexer and the assigner come in one delivery) it is read with no restart: a module only the placement names is + // said as placed by one source only. + pb := startBed(t, delivery.PlacementBucket) + pdone := pb.start(t) + select { + case err := <-pdone: + t.Fatalf("without %s the assigner stopped: %v", delivery.PlacementBucket, err) + case <-time.After(2 * time.Second): + } + placement, err := pb.js.CreateKeyValue(pb.ctx, jetstream.KeyValueConfig{Bucket: delivery.PlacementBucket, History: 5}) + if err != nil { + t.Fatal(err) + } + pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Modules: []delivery.PlacedModule{{Module: "web"}}}) + plkey, _ := delivery.PlacementKey("n1") + if _, err := placement.Put(pb.ctx, plkey, pl); err != nil { + t.Fatal(err) + } + for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(200 * time.Millisecond) { + if c, ok := pb.conditionsSaid(t)[delivery.AssignerModule+".placement.n1.web"]; ok && c.State == asks.StateRaised { + break + } + if time.Now().After(deadline) { + t.Fatal("the placement state made after the start was not read") + } + } } // putJSON writes a value into one of the bed's buckets. @@ -406,6 +437,32 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) { if !cleared { t.Errorf("the condition of web on neither list on g14 was not cleared once g14 got its want") } + + // Issue 1126 (ADR 0547 §2): the catalogue's placement reaches the indexer's state naming web on every node the + // assignments name it on. Each node then composes its own: its want names the build alone (Self, no composition), and + // its own want says it composes its own, which the node agent reads (mesh-host's TestNats...ComposesAndAppliesItsOwn). + for _, node := range []string{"ace", "g14", "shanks"} { + pl, err := delivery.EncodePlacement(delivery.Placement{Node: node, Modules: []delivery.PlacedModule{{Module: "web"}}}) + if err != nil { + t.Fatal(err) + } + pkey, _ := delivery.PlacementKey(node) + b.putJSON(t, delivery.PlacementBucket, pkey, pl) + } + if w := b.wantNow(t, "ace", "web", "ace's want once placed: the build alone, composed by the node", func(w delivery.Want) bool { return w.Self }); w.Composed != "" || w.FP != fNew { + t.Errorf("ace's want once placed = %+v", w) + } + nkey, _ := delivery.WantNodeKey("ace") + var self bool + for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && !self; time.Sleep(100 * time.Millisecond) { + if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, nkey); err == nil { + nw, err := delivery.DecodeNodeWant(e.Value()) + self = err == nil && nw.Self + } + } + if !self { + t.Error("ace's own want does not say it composes its own once placed") + } } // 1297/B9 (the assigner's half, ADR 0528 clause 6): started whole, the assigner holds a fingerprint whose health check diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go index 3ba0133..a3d582b 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go @@ -59,6 +59,7 @@ var readBuckets = []string{ delivery.CandidatesBucket, delivery.ManifestsBucket, delivery.ModulesBucket, + delivery.PlacementBucket, delivery.PresenceBucket, delivery.HaveBucket, delivery.StepsBucket(goBuildModule), @@ -98,6 +99,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err) continue } + if name == delivery.PlacementBucket && err != nil { + // The indexer's placement state comes with the second milestone (ADR 0549 §2): without it the controller's + // assignments alone name where modules run, as before any placement was read (ADR 0549 §5). + logf("the bucket %s cannot be opened (%v): the controller's assignments alone are read until the assigner starts again", name, err) + continue + } if err != nil { return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err) } @@ -151,7 +158,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err) } logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs)) - return follow(ctx, js, kvs, sayVia(ctx, js), emitting{said}, logger) + return follow(ctx, js, conn, kvs, sayVia(ctx, js), emitting{said}, logger) } // emitting is the process connection as the puts' caller: its calls, and its events with no options. @@ -169,6 +176,7 @@ var watched = []string{ delivery.BuilderBuildsBucket, delivery.CandidatesBucket, delivery.ModulesBucket, + delivery.PlacementBucket, delivery.PresenceBucket, delivery.HaveBucket, delivery.AssignerAsksBucket, @@ -179,7 +187,7 @@ var watched = []string{ // follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after // each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends. -func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(asks.Condition) error, bus caller, logger *log.Logger) error { +func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, bus caller, logger *log.Logger) error { w := newWorld() dirty := make(chan struct{}, 1) wake := func() { @@ -263,7 +271,12 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a } logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket) } + started := 0 for _, name := range watched { + if kvs[name] == nil && name == delivery.PlacementBucket { + continue // not on the bus at this start (open said so) + } + started++ var opts []jetstream.WatchOpt if name == delivery.HaveBucket { opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back @@ -275,7 +288,20 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil { return err } - for range len(watched) + 1 { + // The indexer's placement state arrives with the same delivery as this assigner (ADR 0549 §2): when it is not on the + // bus at this start it is bound once the bus says it was created, and watched from then on, with no restart (issue + // 1790's kind). Until then the controller's assignments alone name where modules run (ADR 0549 §5). + placementAppeared := make(chan jetstream.KeyValue, 1) + if kvs[delivery.PlacementBucket] == nil && js != nil { + process.AwaitBucket(ctx, js, nc, delivery.PlacementBucket, "the controller's assignments alone are read", + logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- kv }) + } + takePlacement := func(kv jetstream.KeyValue) error { + kvs[delivery.PlacementBucket] = kv + logger.Printf("the bucket %s is open: placement is read beside the controller's assignments", delivery.PlacementBucket) + return watch(delivery.PlacementBucket) + } + for range started + 1 { select { case <-ctx.Done(): return nil @@ -297,6 +323,10 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a case err := <-errs: return err case <-settled: + case kv := <-placementAppeared: + if err := takePlacement(kv); err != nil { + return err + } case <-dirty: reopenPuts(p) p.pass(ctx) diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go index 75d7411..dc9bc87 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go @@ -94,7 +94,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) { kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}} kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote - go follow(ctx, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0)) + go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0)) select { case key := <-wrote: if key != "want.ace.node" { diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go index 45c3e77..00cb640 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go @@ -60,7 +60,23 @@ func (p *passer) pass(ctx context.Context) { // controller has composed it and again whenever that digest changes (ADR 0539 §6). func (p *passer) nodeWants(ctx context.Context) { written := p.w.read(delivery.AssignerWantsBucket) + placed := p.w.placedNodes() + // A node the catalogue places composes its own composition (ADR 0547 §2): its own want names none. + for node := range placed { + key, err := delivery.WantNodeKey(node) + if err != nil { + continue + } + if before, err := delivery.DecodeNodeWant(written[key]); err == nil && before.Self { + continue + } + want := delivery.NodeWant{Self: true, Words: "the node composes its own composition (ADR 0547 §2)"} + p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeNodeWant(want) }) + } for node, d := range p.w.nodeComposed() { + if placed[node] { + continue + } key, err := delivery.WantNodeKey(node) if err != nil { continue @@ -97,6 +113,8 @@ func (p *passer) wants(ctx context.Context) { } } wanted := p.w.wantsOf() + frozen := p.sayPlacement(p.w.placementDiffers()) + placed := p.w.placedNodes() puts := p.putsByModule() p.staleAsks(ctx, trunk) names := map[string]bool{} @@ -119,7 +137,14 @@ func (p *passer) wants(ctx context.Context) { fp := trunk[name].FP2 name := name m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name], - Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) }, + Composed: func(node, fp string) (delivery.ComposedDigest, bool) { + if placed[node] { + // The node composes its own (ADR 0547 §2, §4): no composition of the controller's is waited for, the want + // names the build alone, and the node judges the move itself. + return delivery.ComposedDigest{Judgement: delivery.JudgedNeither, Words: "the node composes its own"}, true + } + return p.w.composed(node, name, fp) + }, GrantedAt: func(fp string) bool { key, err := delivery.ControllerGrantKey(name, fp) return err == nil && p.w.has(delivery.ControllerGrantsBucket, key) @@ -153,6 +178,7 @@ func (p *passer) wants(ctx context.Context) { last[node] = p.w.lastHealthy(node, name, fp) } out := assign.Wants(m, nodes, wanted[name], last) + p.yeses(ctx, name, m, placed, wanted[name], &out) gatherNeither(neither, name, out) leftNeither(left, m, wanted[name], out) p.sayNoWant(name, m.Nodes, wanted[name], out) @@ -183,6 +209,9 @@ func (p *passer) wants(ctx context.Context) { p.recordAsk(ctx, name, fp, need) } for node, want := range out.Wants { + if frozen[node+"."+name] { + continue // one source alone names it (ADR 0549 §5): no want, and the one written before stays + } key, err := delivery.WantKey(node, name) if err != nil { p.logf("no want of %s on %s is written: %v", name, node, err) @@ -728,3 +757,83 @@ func (p *passer) backupAsk(ctx context.Context, key, node, module string, want d } p.said[c.Key] = true } + +// sayPlacement raises, once, the condition of every pair the controller's assignments and the catalogue's placement do not +// both name, clears it once they agree, and answers those pairs by node and module (ADR 0549 §5): the pass writes no want +// for them and keeps the one written before. A say that fails is made again at the next pass. +func (p *passer) sayPlacement(differ []assign.Pair) map[string]bool { + frozen := map[string]bool{} + now := map[string]bool{} + for _, d := range differ { + frozen[d.Node+"."+d.Module] = true + c := assign.PlacementCondition(d, true) + now[c.Key] = true + if p.said[c.Key+".raised"] { + continue + } + p.logf("%s on %s is named by %s alone: no want is written for it there until both sources name it (ADR 0549 §5)", d.Module, d.Node, d.Only) + if err := p.say(c); err != nil { + p.logf("the placement condition of %s on %s could not be said, and is said at the next pass: %v", d.Module, d.Node, err) + continue + } + p.said[c.Key+".raised"] = true + } + prefix := delivery.AssignerModule + ".placement." + for key := range p.said { + c, raised := strings.CutSuffix(key, ".raised") + if !raised || !strings.HasPrefix(c, prefix) || now[c] { + continue + } + node, module, _ := strings.Cut(strings.TrimPrefix(c, prefix), ".") + if err := p.say(assign.PlacementCondition(assign.Pair{Node: node, Module: module}, false)); err != nil { + p.logf("the placement condition of %s on %s could not be cleared, and is cleared at the next pass: %v", module, node, err) + continue + } + delete(p.said, key) + } + return frozen +} + +// yeses writes a placed node's wants naming the build alone, and sees to the moves a node that composes its own judged a +// person must allow (ADR 0547 §3): for a have of phase +// waiting-yes it raises the record ask once, naming the composition waiting and the node's judgement, and on a yes writes +// the want again naming it; a yes once written is kept while the want stays at its build, so the node is not asked again. +func (p *passer) yeses(ctx context.Context, module string, m assign.Module, placed map[string]bool, before map[string]delivery.Want, out *assign.Outcome) { + for _, node := range m.Nodes { + if !placed[node] { + continue + } + was, had := before[node] + w, written := out.Wants[node] + if !written { + if !had || was.Removed { + continue + } + w = was + } + if w.Removed { + continue + } + yes := "" + if had && was.FP == w.FP { + yes = was.Yes + } + key, err := delivery.HaveKey(node, module) + if err != nil { + continue + } + if h, err := delivery.DecodeHave(p.w.value(delivery.HaveBucket, key)); err == nil && h.Phase == delivery.PhaseWaitingYes && h.Waits != yes { + if m.Allowed != nil && m.Allowed(node, h.Waits) { + yes = h.Waits + } else { + p.recordAsk(ctx, module, w.FP, assign.RecordNeed{Node: node, Digest: h.Waits, Judgement: h.Judgement, Words: h.Detail}) + } + } + // A want carried over from before the node composed its own may still name the controller's composition; a placed + // node's want names the build alone (ADR 0547 §2). + if yes != w.Yes || w.Composed != "" || !w.Self { + w.Yes, w.Composed, w.Self = yes, "", true + out.Wants[node] = w + } + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go index 7a89fef..15fbde1 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go @@ -342,6 +342,9 @@ func putOfTheBus(t *testing.T) { // again after a say that failed, and clears it once when the module has a want again; a module on neither list is also written to // the journal, so it is loud with no operator channel. func TestAModuleOnNeitherListIsRaisedOnceAndClearedOnce(t *testing.T) { + // A pair only one source names is a condition of the same kind (issue 1126): folded here to hold the count. + aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t) + t.Run("gathers", testThePassGathersTheModulesOnNeitherList) p := newPasser(buckets{}, time.Date(2026, 10, 25, 15, 0, 0, 0, time.UTC)) var said []asks.Condition failing := true @@ -391,7 +394,7 @@ func TestAModuleOnNeitherListIsRaisedOnceAndClearedOnce(t *testing.T) { } // 1945/AC21: the pass gathers the modules on neither list of every module's outcome, keyed by node and module. -func TestThePassGathersTheModulesOnNeitherList(t *testing.T) { +func testThePassGathersTheModulesOnNeitherList(t *testing.T) { got := map[string]neitherOne{} gatherNeither(got, "screen-lock", assign.Outcome{Neither: map[string]assign.Holding{"g14": {Rule: delivery.RuleGrant, Words: "w"}}}) gatherNeither(got, "rofi", assign.Outcome{}) @@ -449,6 +452,7 @@ func TestThePassGathersTheModulesOnNeitherList(t *testing.T) { // nothing it said, so it clears, once, the neither condition of every assigned module not on neither list and the held condition of // every module that is not held. Clearing a condition that is not open is quiet at the router. func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) { + t.Run("a held module", testAHeldModuleRaisesNoNeitherAndClearsAStandingOne) wantsKV := &memKV{values: map[string][]byte{}} p := newPasser(buckets{delivery.AssignerWantsBucket: wantsKV, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now()) var said []asks.Condition @@ -486,7 +490,7 @@ func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) { // 1945/AC24: given a module held at F under rule 2 on a node that runs it with no want, when the assigner passes, then it raises no // neither condition for it, a neither condition this process raised for it before is cleared, and so is one a previous process // raised (the first pass after a start), each once. -func TestAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) { +func testAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) { setup := func() (*passer, *[]asks.Condition) { p := newPasser(buckets{delivery.AssignerWantsBucket: &memKV{values: map[string][]byte{}}, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now()) said := &[]asks.Condition{} @@ -702,6 +706,8 @@ func mustKey(key string, err error) string { // (ADR 0545). That record ask is not about the trunk's fingerprint: it survives every pass, so the operator is asked, and on a // yes the want names the revised digest. func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing.T) { + // A placed node's yes comes from its have (issue 1126): folded here to hold the count. + aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t) fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64) revised, was := "sha256:"+strings.Repeat("e", 64), "sha256:"+strings.Repeat("d", 64) asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}} @@ -750,3 +756,133 @@ func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing t.Error("the answered record ask was taken away") } } + +// 1126 (ADR 0549 §5): until the third milestone a pair the controller's assignments and the catalogue's placement do not +// both name gets no want and one condition, raised once and cleared once both name it; before any placement was read the +// assignments are the one source and nothing is raised. +func aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) { + p := newPasser(buckets{}, time.Date(2026, 10, 26, 18, 0, 0, 0, time.UTC)) + var said []asks.Condition + p.say = func(c asks.Condition) error { said = append(said, c); return nil } + assigned := func(node, module string) { + raw, _ := json.Marshal(map[string]string{"node": node, "module": module}) + p.w.put(provider.AssignmentsBucket, node+"."+module, raw) + } + place := func(node string, modules ...string) { + pl := delivery.Placement{Node: node} + for _, m := range modules { + pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m}) + } + raw, err := delivery.EncodePlacement(pl) + if err != nil { + t.Fatal(err) + } + key, _ := delivery.PlacementKey(node) + p.w.put(delivery.PlacementBucket, key, raw) + } + assigned("g14", "a") + assigned("g14", "b") + if got, _ := p.w.assigned(); !slices.Equal(got["g14"], []string{"a", "b"}) || len(p.sayPlacement(p.w.placementDiffers())) != 0 || len(said) != 0 { + t.Fatalf("before any placement: assigned %v, said %v", got, said) + } + none := func(context.Context, string) ([]byte, error) { return nil, errNotThere } + if m := p.w.mesh(context.Background(), none); m.Input != nil || m.InputUnread != "" { + t.Errorf("before any placement the check composes over %+v (%q)", m.Input, m.InputUnread) + } + place("g14", "a", "c") + got, _ := p.w.assigned() + frozen := p.sayPlacement(p.w.placementDiffers()) + p.sayPlacement(p.w.placementDiffers()) + if !slices.Equal(got["g14"], []string{"a"}) || !frozen["g14.b"] || !frozen["g14.c"] || frozen["g14.a"] || len(said) != 2 || + said[0].State != asks.StateRaised || !strings.Contains(said[0].Explanation, "the controller's assignments names b on g14") || + !strings.Contains(said[1].Explanation, "the catalogue's placement names c on g14") { + t.Fatalf("one source alone: assigned %v, frozen %v, said %+v", got, frozen, said) + } + place("g14", "a", "b", "c") + assigned("g14", "c") + if frozen := p.sayPlacement(p.w.placementDiffers()); len(frozen) != 0 || len(said) != 4 || said[2].State != asks.StateCleared || said[3].State != asks.StateCleared { + t.Errorf("both agree again: frozen %v, said %+v", frozen, said) + } + + // 1126 (ADR 0547 §2): placed, the check composes over the input the nodes compose from (frombus), once each placed node + // wrote its record; until then it says why and composes the assignments alone. + if m := p.w.mesh(context.Background(), none); m.Input != nil || !strings.Contains(m.InputUnread, "g14") { + t.Errorf("a placed node with no record: input %+v, unread %q", m.Input, m.InputUnread) + } + rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14", TunnelKey: "k"}}) + p.w.put(delivery.HaveBucket, "g14.node", rec) + m := p.w.mesh(context.Background(), none) + if m.Input == nil || len(m.Input.Nodes) != 1 || !slices.Equal(m.Input.Nodes[0].Assigned, []string{"a", "b", "c"}) || m.Input.Nodes[0].Place.Key != "k" { + t.Errorf("the check's input from the bus: %+v (%q)", m.Input, m.InputUnread) + } +} + +// 1126 (ADR 0547 §2 to §4): a node the catalogue places composes its own: its wants name the build alone and its own want +// names no composition; a have waiting for a yes raises the record ask once, and the yes is written into the want and +// kept there while the want stays at its build. +func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) { + fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64) + waiting, was := "sha256:"+strings.Repeat("e", 64), "sha256:"+strings.Repeat("d", 64) + asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}} + p := newPasser(buckets{delivery.AssignerAsksBucket: asksKV, delivery.AssignerWantsBucket: wantsKV}, time.Now()) + p.logf = t.Logf + put := func(bucket, key string, v []byte) { p.w.put(bucket, key, v) } + mod, _ := delivery.EncodeModule(delivery.Module{FP2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath}) + put(delivery.ModulesBucket, "web", mod) // the trunk's build is not done: rule 1 holds g14 at fpOld + put(provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-1"}`)) + pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}}) + put(delivery.PlacementBucket, "placement.g14", pl) + on, _ := delivery.EncodePresence(delivery.Presence{Online: true}) + put(delivery.PresenceBucket, "g14", on) + have, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Composed: was, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + put(delivery.HaveBucket, "g14.web", have) + wkey, _ := delivery.WantKey("g14", "web") + want, _ := delivery.EncodeWant(delivery.Want{FP: fpOld, Composed: was}) + wantsKV.values[wkey] = want + put(delivery.AssignerWantsBucket, wkey, want) + sync := func() { + for k, v := range wantsKV.values { + put(delivery.AssignerWantsBucket, k, v) + } + } + + p.wants(context.Background()) + p.nodeWants(context.Background()) + sync() + got, err := delivery.DecodeWant(wantsKV.values[wkey]) + if err != nil || got.FP != fpOld || got.Composed != "" || !got.Self { + t.Fatalf("the placed node's want is %+v (%v), want it at %s, the node composing it", got, err, fpOld) + } + nkey, _ := delivery.WantNodeKey("g14") + if nw, err := delivery.DecodeNodeWant(wantsKV.values[nkey]); err != nil || nw.Composed != "" || !nw.Self { + t.Fatalf("the placed node's own want is %+v (%v), want one naming no composition", nw, err) + } + + waits, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Composed: was, Phase: delivery.PhaseWaitingYes, Health: delivery.HealthOK, + Judgement: delivery.JudgedRecreates, Waits: waiting, Detail: "it recreates web.server"}) + put(delivery.HaveBucket, "g14.web", waits) + akey, _ := delivery.RecordAskKey("g14", "web", waiting) + p.wants(context.Background()) + if _, there := asksKV.values[akey]; !there { + t.Fatal("a have waiting for a yes raised no record ask") + } + a, err := delivery.DecodeAssignerAsk(akey, asksKV.values[akey]) + if err != nil { + t.Fatal(err) + } + put(delivery.AssignerAsksBucket, akey, asksKV.values[akey]) + yes, _ := delivery.EncodeAnswer(delivery.Answer{Option: assign.OptionYes, Warrant: a.Ask.ID + "@2026-10-26T19:00:00Z", Words: "the operator"}) + put(delivery.AnswersBucket, akey, yes) + p.wants(context.Background()) + sync() + if got, err = delivery.DecodeWant(wantsKV.values[wkey]); err != nil || got.Yes != waiting || got.FP != fpOld { + t.Fatalf("after the yes the want is %+v (%v), want it to name the yes %s", got, err, waiting) + } + // The node applied it: its have runs the composition; the yes stays while the want stays at its build. + ran, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Composed: waiting, Phase: delivery.PhaseRunning, Health: delivery.HealthOK}) + put(delivery.HaveBucket, "g14.web", ran) + p.wants(context.Background()) + if got, _ = delivery.DecodeWant(wantsKV.values[wkey]); got.Yes != waiting { + t.Errorf("a pass after the move dropped the yes: %+v", got) + } +} diff --git a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go index 7fbb11a..22fb1d4 100644 --- a/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go +++ b/modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go @@ -14,6 +14,8 @@ import ( "github.com/nats-io/nats.go/jetstream" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" + "git.novox.be/novox/mesh-sdk/go/compose/frombus" "git.novox.be/novox/mesh-sdk/go/delivery" "git.novox.be/novox/mesh-sdk/go/provider" @@ -162,8 +164,56 @@ func (w *world) manifest(ctx context.Context, get getter, digest string) parsed return p } -// assigned is each node's modules with an open assignment, read from the controller's assignments, in name order. +// assigned is each node's modules both the controller's assignments and the indexer's placement name, in name order +// (ADR 0549 §5; the assignments alone before any placement was read), with every open assignment by node and module. func (w *world) assigned() (map[string][]string, map[string]assignment) { + byAssignment, all := w.assignedByController() + placed, seeded := w.placed() + both, _ := assign.Placed(byAssignment, placed, seeded) + return both, all +} + +// placementDiffers is every pair the controller's assignments and the indexer's placement do not both name. +func (w *world) placementDiffers() []assign.Pair { + byAssignment, _ := w.assignedByController() + placed, seeded := w.placed() + _, differ := assign.Placed(byAssignment, placed, seeded) + return differ +} + +// placed is each node's modules in the indexer's placement (ADR 0549 §2), and whether any placement key was read: a key +// that does not read places nothing on its node and is said by the indexer's own read tool. +func (w *world) placed() (map[string][]string, bool) { + out := map[string][]string{} + values := w.read(delivery.PlacementBucket) + for key, value := range values { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + p, err := delivery.DecodePlacement(value) + if err != nil || p.Node != node { + continue + } + out[node] = p.Assigned() + } + return out, len(values) > 0 +} + +// placedNodes is every node the indexer's placement names (ADR 0549 §2): those compose their own modules (ADR 0547 §2), +// so their wants name the build alone and the seventh rule does not hold them. +func (w *world) placedNodes() map[string]bool { + placed, _ := w.placed() + out := make(map[string]bool, len(placed)) + for node := range placed { + out[node] = true + } + return out +} + +// assignedByController is each node's modules with an open assignment, read from the controller's assignments, in +// name order. +func (w *world) assignedByController() (map[string][]string, map[string]assignment) { out, all := map[string][]string{}, map[string]assignment{} for _, value := range w.read(provider.AssignmentsBucket) { var a assignment @@ -210,9 +260,86 @@ func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh { } m.Manifests[name] = p.manifest } + // Once the catalogue places the nodes, the check composes over the input they compose from (ADR 0547 §2, ADR 0549 + // §3): one composer's input for the check and the nodes. An input that cannot be read is said, and the nodes and their + // assignments alone are composed, as before any placement. + if _, seeded := w.placed(); seeded { + in, err := frombus.Input(ctx, worldReader{w: w, manifests: manifests}, frombus.Mesh{}, frombus.Scope{Every: true}) + if err != nil { + m.InputUnread = err.Error() + } else { + m.Input = &in + } + } return m } +// worldReader is what the assigner watched as frombus reads the bus: the indexer's placement and modules, each node's +// record in its own have key, the manifests by digest; the check reads no config value, own secret or credential (the +// composition it asks resolves, and stands in for what it does not carry). +type worldReader struct { + w *world + manifests getter +} + +func (r worldReader) Placements(context.Context) (map[string]delivery.Placement, error) { + out := map[string]delivery.Placement{} + for key, value := range r.w.read(delivery.PlacementBucket) { + node, err := delivery.ParsePlacementKey(key) + if err != nil { + continue + } + if p, err := delivery.DecodePlacement(value); err == nil && p.Node == node { + out[node] = p + } + } + return out, nil +} + +func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, error) { + out := map[string]delivery.NodeRecord{} + for key, value := range r.w.read(delivery.HaveBucket) { + node, _, isNode, err := delivery.ParseHaveKey(key) + if err != nil || !isNode { + continue + } + if h, err := delivery.DecodeHaveNode(value); err == nil && h.Record != nil { + out[node] = *h.Record + } + } + return out, nil +} + +func (r worldReader) Modules(context.Context) (map[string]delivery.Module, error) { + return r.w.trunk(), nil +} + +func (r worldReader) Manifest(ctx context.Context, digest string) ([]byte, error) { + key, err := delivery.ManifestsKey(digest) + if err != nil { + return nil, err + } + raw, err := r.manifests(ctx, key) + if err != nil { + return nil, err + } + m, err := delivery.DecodeManifest(raw) + if err != nil { + return nil, err + } + return []byte(m.JSON), nil +} + +func (worldReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil } + +func (worldReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) { + return nil, nil +} + +func (worldReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) { + return nil, nil, nil +} + // change is a candidate's change for the composition check, or why the composition waits for it. func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) { ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{}, diff --git a/modules/mesh-delivery-assigner/go.mod b/modules/mesh-delivery-assigner/go.mod index df8760f..9476f6d 100644 --- a/modules/mesh-delivery-assigner/go.mod +++ b/modules/mesh-delivery-assigner/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-assigner go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-assigner/go.sum b/modules/mesh-delivery-assigner/go.sum index 8492faa..e8660dd 100644 --- a/modules/mesh-delivery-assigner/go.sum +++ b/modules/mesh-delivery-assigner/go.sum @@ -1,13 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c h1:pY2YpHSSsfDPBrOEiWTNSIPn0zdnAayRmtmmyQ3hCHY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025195051-20a6dc4920f8 h1:jA7V7hlfJAigVUvJDpKwfgd9ylQ9bqppg3rK9XgLWoA= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025195051-20a6dc4920f8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8 h1:vhYk/iv43u2WGSTrFyQGDvGBxLaRITgMAiiI9DjuA10= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 h1:gE80bR0tPME7MTMBLcdP62v+F/9soVXmdcBYNMOye44= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-assigner/internal/assign/compose.go b/modules/mesh-delivery-assigner/internal/assign/compose.go index b1f26e4..cb788b4 100644 --- a/modules/mesh-delivery-assigner/internal/assign/compose.go +++ b/modules/mesh-delivery-assigner/internal/assign/compose.go @@ -35,6 +35,13 @@ type Mesh struct { // Digests is every trunk module's manifest digest in the indexer's manifests bucket, readable or not: what a // composer at another mesh-sdk reads (issue 1622). Digests map[string]string + // Input is the rest of the mesh as the nodes compose from it, filled by mesh-sdk's frombus from the catalogue's + // placement and the nodes' records (novox/hq ADR 0547 §2, ADR 0549 §3, issue 1126), so the check and the nodes share + // one composer's input; nil before any placement is read, when the nodes and their assignments alone are composed. + Input *compose.Input + // InputUnread says why that input could not be read from the bus, when it could not: the check then composes the + // nodes and their assignments alone, and says so in its warnings. + InputUnread string } // Change is a group's change as the indexer's candidate holds it: the manifest of every module it moves or adds, those diff --git a/modules/mesh-delivery-assigner/internal/assign/composer.go b/modules/mesh-delivery-assigner/internal/assign/composer.go index b8ccb02..8c60ce8 100644 --- a/modules/mesh-delivery-assigner/internal/assign/composer.go +++ b/modules/mesh-delivery-assigner/internal/assign/composer.go @@ -42,8 +42,15 @@ func HeadJob(m Mesh, c Change) (delivery.ComposeJob, error) { for _, name := range c.Removed { delete(change, name) } - req := composer.Request{V: composer.Version, Input: plainInput(m.Assigned), Depth: composer.DepthResolve, - HoldingsFromClaims: true, Machines: Touched(m, c), + in, fromClaims := plainInput(m.Assigned), true + if m.Input != nil { + // The input the nodes compose from, without its catalogue: the cases carry the manifests (issue 1126). + in = *m.Input + in.Catalogue = nil + fromClaims = len(in.Holdings) == 0 + } + req := composer.Request{V: composer.Version, Input: in, Depth: composer.DepthResolve, + HoldingsFromClaims: fromClaims, Machines: Touched(m, c), Cases: []composer.Case{{Name: CaseTrunk, Manifests: trunk}, {Name: CaseChange, Manifests: change}}} return job(SDKRepository, c.SDKHead, SDKDir, req, "the mesh-sdk head of "+c.Group) } @@ -195,6 +202,10 @@ func FromComposer(m Mesh, c Change, head Ran, readers []Reader, took time.Durati readerFails, readerWarnings := judgeReaders(c, change, readers) fails = append(fails, readerFails...) v.Warnings = append(v.Warnings, readerWarnings...) + if m.InputUnread != "" { + v.Warnings = append(v.Warnings, "composed over the nodes and their assignments alone: the input the nodes compose from "+ + "cannot be read from the bus ("+m.InputUnread+")") + } return finish(v, m, c, machines, fails) } diff --git a/modules/mesh-delivery-assigner/internal/assign/composer_test.go b/modules/mesh-delivery-assigner/internal/assign/composer_test.go index 4b709c8..2bc3258 100644 --- a/modules/mesh-delivery-assigner/internal/assign/composer_test.go +++ b/modules/mesh-delivery-assigner/internal/assign/composer_test.go @@ -8,6 +8,7 @@ import ( "time" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/compose" "git.novox.be/novox/mesh-sdk/go/composer" "git.novox.be/novox/mesh-sdk/go/delivery" ) @@ -100,6 +101,25 @@ func TestAGroupWithAnSDKMemberAsksTheComposerAtItsHead(t *testing.T) { if !slices.Equal(req.Machines, []string{"g14"}) || req.Depth != composer.DepthResolve || !req.HoldingsFromClaims { t.Errorf("request = machines %v, depth %q, holdings from claims %v", req.Machines, req.Depth, req.HoldingsFromClaims) } + + // 1126 (ADR 0547 §2, ADR 0549 §3): once the catalogue places the nodes, the check composes over the input the nodes + // compose from (frombus), its catalogue left to the cases, and the seats held as placement says. + m := digestedMesh() + m.Input = &compose.Input{Catalogue: map[string]catalogue.Manifest{"web": {Module: "web"}}, + Nodes: []compose.Node{{Name: "g14", Assigned: []string{"web"}, Place: compose.Place{Endpoint: "g14.example:51820"}}}, + Holdings: []catalogue.Held{{Claim: "web-seat", Scope: "mesh", Node: "g14", Module: "web"}}} + placedJob, err := HeadJob(m, c) + if err != nil { + t.Fatal(err) + } + placed, err := composer.Decode(placedJob.Request) + if err != nil { + t.Fatal(err) + } + if placed.Input.Catalogue != nil || len(placed.Input.Nodes) != 1 || placed.Input.Nodes[0].Place.Endpoint != "g14.example:51820" || + len(placed.Input.Holdings) != 1 || placed.HoldingsFromClaims { + t.Errorf("the placed check's request = %+v, holdings from claims %v", placed.Input, placed.HoldingsFromClaims) + } } // 1622/AC2, AC3: a catalogue member using a field only the SDK head reads passes with the head's composer, and the diff --git a/modules/mesh-delivery-assigner/internal/assign/placement.go b/modules/mesh-delivery-assigner/internal/assign/placement.go new file mode 100644 index 0000000..9856b12 --- /dev/null +++ b/modules/mesh-delivery-assigner/internal/assign/placement.go @@ -0,0 +1,99 @@ +package assign + +import ( + "fmt" + "sort" + + "git.novox.be/novox/mesh-sdk/go/asks" + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// Until ADR 0547's third milestone two sources name which module runs on which node: the controller's assignments and +// the indexer's placement, read from the catalogue (ADR 0549 §5). The assigner writes wants only for a pair both name: +// for a pair one source names alone it writes no want and touches none it wrote before, so nothing moves and nothing is +// removed on one source's word, and it raises a condition naming the pair until both agree. Before the first placement +// file reaches the indexer (no placement key at all) the controller's assignments are the one source, as before. + +// Pair is a module on a node that one source names and the other does not; Only says which: OnlyAssignments or +// OnlyPlacement. +type Pair struct { + Node, Module, Only string +} + +// The source that names a pair alone. +const ( + OnlyAssignments = "the controller's assignments" + OnlyPlacement = "the catalogue's placement" +) + +// Placed is, by node, the modules both sources name, in name order, and every pair one names alone, by node then +// module. With no placement read yet (seeded false) it is the assignments, and no pair differs. +func Placed(assigned, placed map[string][]string, seeded bool) (map[string][]string, []Pair) { + if !seeded { + return assigned, nil + } + set := func(m map[string][]string) map[string]bool { + out := map[string]bool{} + for node, modules := range m { + for _, module := range modules { + out[node+"\x00"+module] = true + } + } + return out + } + inAssigned, inPlaced := set(assigned), set(placed) + both := map[string][]string{} + var differ []Pair + for node, modules := range assigned { + for _, module := range modules { + if inPlaced[node+"\x00"+module] { + both[node] = append(both[node], module) + } else { + differ = append(differ, Pair{Node: node, Module: module, Only: OnlyAssignments}) + } + } + } + for node, modules := range placed { + for _, module := range modules { + if !inAssigned[node+"\x00"+module] { + differ = append(differ, Pair{Node: node, Module: module, Only: OnlyPlacement}) + } + } + } + for _, modules := range both { + sort.Strings(modules) + } + sort.Slice(differ, func(i, j int) bool { + if differ[i].Node != differ[j].Node { + return differ[i].Node < differ[j].Node + } + return differ[i].Module < differ[j].Module + }) + return both, differ +} + +// PlacementCondition is the condition of a pair the two sources do not both name: raised while they differ, cleared once +// both name it or neither does. +func PlacementCondition(p Pair, raised bool) asks.Condition { + key := delivery.AssignerModule + ".placement." + p.Node + "." + p.Module + subject := asks.Scope{Scope: "module", ID: p.Module, Machine: p.Node} + if !raised { + return asks.Condition{Key: key, Subject: subject, State: asks.StateCleared, + Resolved: fmt.Sprintf("the controller's assignments and the catalogue's placement agree on %s on %s.", p.Module, p.Node)} + } + headline := fmt.Sprintf("%s on %s is placed by one source only", p.Module, p.Node) + if n := len([]rune(headline)); n > asks.HeadlineLength { + headline = string([]rune(headline)[:asks.HeadlineLength]) + } + var needs string + if p.Only == OnlyAssignments { + needs = fmt.Sprintf("add %s to the catalogue's placement file of %s, or unassign it at the controller", p.Module, p.Node) + } else { + needs = fmt.Sprintf("assign %s to %s at the controller, as the merged placement file says", p.Module, p.Node) + } + return asks.Condition{Key: key, Severity: asks.SeverityUrgent, Subject: subject, State: asks.StateRaised, + Summary: headline, Headline: headline, Needs: needs, + Explanation: fmt.Sprintf("%s%s. %s names %s on %s and the other source does not. Until ADR 0547's third milestone both must "+ + "name a module on a node (ADR 0549 §5): the assigner writes no want for it there and leaves the one it wrote before, so "+ + "nothing moves or is removed. This clears once both name it or neither does.", asks.NeedsYou, needs, p.Only, p.Module, p.Node)} +} diff --git a/modules/mesh-delivery-assigner/manifest_test.go b/modules/mesh-delivery-assigner/manifest_test.go index cf387b7..3c6e8a9 100644 --- a/modules/mesh-delivery-assigner/manifest_test.go +++ b/modules/mesh-delivery-assigner/manifest_test.go @@ -34,7 +34,7 @@ func TestManifest(t *testing.T) { slices.Sort(reads) want := []string{"mesh-build-go.steps", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", - "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"} + "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"} if !reflect.DeepEqual(reads, want) { t.Errorf("reads = %v, want exactly %v", reads, want) } diff --git a/modules/mesh-delivery-assigner/module.json b/modules/mesh-delivery-assigner/module.json index 3c31dc8..a979017 100644 --- a/modules/mesh-delivery-assigner/module.json +++ b/modules/mesh-delivery-assigner/module.json @@ -12,6 +12,7 @@ "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have" diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go index 9310a73..51fe90a 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go @@ -91,6 +91,28 @@ func treesVia(ctx context.Context, call caller) bridge.Trees { } } +// listedVia asks the forge for the files directly under one directory of a commit through call: gitea_commit_trees with +// listed (issue 1126), answer {listed: {directory: {name: blob id}}}. +func listedVia(ctx context.Context, call caller) bridge.Listed { + return func(owner, repo, sha, dir string) (map[string]string, error) { + raw, err := call(ctx, commitTreesTool, map[string]any{"owner": owner, "repo": repo, "sha": sha, "listed": []string{dir}}) + if err != nil { + return nil, err + } + var a struct { + Listed map[string]map[string]string `json:"listed"` + } + if err := json.Unmarshal(raw, &a); err != nil { + return nil, fmt.Errorf("the answer of %s is not {listed}: %w", commitTreesTool, err) + } + if a.Listed == nil { + // A forge module from before listed answers none: placement cannot be read, never read as empty. + return nil, fmt.Errorf("the answer of %s lists no directory; the forge module does not list yet", commitTreesTool) + } + return a.Listed[dir], nil + } +} + // getFileTool is the forge's tool for one file at a ref (mesh-catalog modules/gitea tools/index.ts gitea_get_file): // arguments owner, repo, path and ref, answer {file: {path, ref, sha, size, content}}, the content decoded to text // (client.ts getFile). A path the forge does not have fails with `Gitea API : 404 ` (client.ts request). diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go index e069a6c..dbdf411 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go @@ -117,8 +117,8 @@ type publisher interface { // handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never // redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server // delivers it again (the server's ack wait is the only clock). -func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, logger *log.Logger) { - msg, err := bridge.Convert(m.Subject(), m.Data(), trees, files) +func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, logger *log.Logger) { + msg, err := bridge.ConvertWith(m.Subject(), m.Data(), trees, files, list) switch { case errors.Is(err, bridge.ErrRetry): logger.Printf("%v; heard again in %s", err, retryDelay) @@ -237,7 +237,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro // Bounded: the turn is held, and no sync or other event goes on without it. hctx, cancel := context.WithTimeout(ctx, syncDeadline) defer cancel() - handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), logger) + handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger) }) }, jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) })) diff --git a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go index 9a16a64..85df582 100644 --- a/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go +++ b/modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go @@ -81,7 +81,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit }) var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{} - handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull" { t.Fatalf("published on %v", pub.subjects) } @@ -97,7 +97,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(map[string]any{"head_sha": "short"})}, &fakePub{} - handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) if len(pub.subjects) != 0 { t.Errorf("a message was written: %v", pub.subjects) } @@ -112,7 +112,7 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5 func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here) var logs bytes.Buffer m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")} - handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0)) + handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0)) if m.acked != 0 || m.termd != 0 { t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd) } @@ -134,7 +134,7 @@ func TestTreesRefusedLeavesTheEventForTenSecondsAndLogs(t *testing.T) { // 1236/ call := func(context.Context, string, any) (json.RawMessage, error) { return nil, errors.New("the tree listing of novox/mesh-delivery was cut") } - handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), log.New(&logs, "", 0)) + handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0)) if len(pub.subjects) != 0 { t.Errorf("a message was written: %v", pub.subjects) } @@ -163,7 +163,7 @@ func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/ } return json.RawMessage(`{"root":"r0","trees":{"":"t0","modules/a":"t1"}}`), nil } - handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), log.New(&logs, "", 0)) + handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0)) if len(keys) == 0 || keys[0] != "gitea.gitea_commit_trees" || slices.Index(keys[1:], "gitea.gitea_commit_trees") >= 0 { t.Fatalf("calls %v, want the trees once and first", keys) } diff --git a/modules/mesh-delivery-forge-bridge/go.mod b/modules/mesh-delivery-forge-bridge/go.mod index 6f902ec..6d3738c 100644 --- a/modules/mesh-delivery-forge-bridge/go.mod +++ b/modules/mesh-delivery-forge-bridge/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-forge-bridge/go.sum b/modules/mesh-delivery-forge-bridge/go.sum index bdc8b83..e8660dd 100644 --- a/modules/mesh-delivery-forge-bridge/go.sum +++ b/modules/mesh-delivery-forge-bridge/go.sum @@ -1,7 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9 h1:Ufmd2cYEXICQ817IGflccP9bexABE4WhIO49HPbaoSo= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go index a6cbc7d..e65f8fd 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go @@ -7,6 +7,7 @@ import ( "encoding/json" "errors" "fmt" + "maps" "path" "regexp" "slices" @@ -88,6 +89,17 @@ func unreadable(subject, why string, args ...any) error { // the trunk asks trees once and becomes a merged message; a merge into any other branch is a closed pull message, as the // pull request left its group. A forge that cannot say is ErrRetry; a merge with no module.json is ErrNoModules. func Convert(subject string, data []byte, trees Trees, files Files) (Message, error) { + return ConvertWith(subject, data, trees, files, nil) +} + +// Listed asks the forge for the files directly under one directory of a commit, by name to blob id (gitea_commit_trees +// with listed); a directory the commit lacks lists none. +type Listed func(owner, repo, sha, dir string) (map[string]string, error) + +// ConvertWith is Convert; with list, a merge of the catalogue repository (delivery.CatalogueRepository) also carries every +// placement file of its merge commit (novox/hq ADR 0549 §2, issue 1126), read whole or not at all: a forge that cannot +// say is ErrRetry, as for the manifests, so the indexer never takes a part of placement for all of it. +func ConvertWith(subject string, data []byte, trees Trees, files Files, list Listed) (Message, error) { var action string switch subject { case UpdatedSubject: @@ -114,7 +126,7 @@ func Convert(subject string, data []byte, trees Trees, files Files) (Message, er return Message{}, unreadable(subject, "no branch") } if subject == MergedSubject && p.Base == Trunk { - return merged(subject, p, trees, files) + return merged(subject, p, trees, files, list) } f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number, Head: p.HeadSHA, Action: action} @@ -227,7 +239,7 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma } // merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once. -func merged(subject string, p pull, trees Trees, files Files) (Message, error) { +func merged(subject string, p pull, trees Trees, files Files, list Listed) (Message, error) { if !sha40.MatchString(p.MergeSHA) { return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA) } @@ -252,8 +264,15 @@ func merged(subject string, p pull, trees Trees, files Files) (Message, error) { if err != nil { return Message{}, unreadable(subject, "%v", err) } - body, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number, - Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks}) + f := delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number, + Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks} + if list != nil && delivery.SameRepository(repository, delivery.CatalogueRepository) { + if f.Placements, err = placementsOf(p.Owner, p.Repo, p.MergeSHA, list, files); err != nil { + return Message{}, fmt.Errorf("%w: %s: the placement files of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err) + } + f.PlacementsRead = true + } + body, err := delivery.EncodeForge(f) if err != nil { return Message{}, unreadable(subject, "%v", err) } @@ -350,3 +369,30 @@ func OpenMessage(answer []byte) (Message, error) { } return Message{Event: delivery.OpenEvent(), Data: body}, nil } + +// placementsOf is every placement file of the catalogue at a commit, by node, its bytes exact: each `.json` directly +// under delivery.PlacementDir. A file whose name is no node is left out and the rest read; the catalogue's check refuses +// such a file before it merges (ADR 0549 §1). Nil when there is none. +func placementsOf(owner, repo, sha string, list Listed, files Files) (map[string]string, error) { + names, err := list(owner, repo, sha, delivery.PlacementDir) + if err != nil { + return nil, err + } + var out map[string]string + for _, name := range slices.Sorted(maps.Keys(names)) { + file := delivery.PlacementDir + "/" + name + node, ok := delivery.PlacementFileNode(file) + if !ok { + continue + } + raw, err := files(owner, repo, file, sha) + if err != nil { + return nil, fmt.Errorf("%s: %w", file, err) + } + if out == nil { + out = map[string]string{} + } + out[node] = string(raw) + } + return out, nil +} diff --git a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go index a5617ec..d4afe54 100644 --- a/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go +++ b/modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go @@ -79,6 +79,47 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1 if want := map[string]map[string]string{"modules/a": {"go.sum": hexSum(sum)}}; !reflect.DeepEqual(f.Locks, want) { t.Errorf("locks = %v, want %v", f.Locks, want) } + if f.Placements != nil || f.PlacementsRead { + t.Errorf("a merge of another repository than the catalogue carries placement: %v", f.Placements) + } + + // Issue 1126 (ADR 0549 §2): a merge of the catalogue carries every placement file of its merge commit, by node; a file + // whose name is no node is left out; a listing or a file the forge cannot give is a retry, never a part. + catalogue := mergedBody(map[string]any{"repo": "mesh-catalog"}) + forge.have["placement/ace.json"] = []byte(`{"v":1,"node":"ace","modules":[]}`) + forge.have["placement/g14.json"] = []byte(`{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`) + listing := map[string]string{"ace.json": "b1", "g14.json": "b2", "README.md": "b3", "a.b.json": "b4"} + var listedAt []string + list := func(owner, repo, sha, dir string) (map[string]string, error) { + listedAt = append(listedAt, owner+"/"+repo+"@"+sha+":"+dir) + return listing, nil + } + msg, err = ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list) + if err != nil { + t.Fatal(err) + } + if f, err = delivery.DecodeForge(msg.Data); err != nil { + t.Fatal(err) + } + want := map[string]string{"ace": `{"v":1,"node":"ace","modules":[]}`, "g14": `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`} + if !f.PlacementsRead || !reflect.DeepEqual(f.Placements, want) || !reflect.DeepEqual(listedAt, []string{"novox/mesh-catalog@" + merge + ":placement"}) { + t.Errorf("placements = %v (read %v), listed at %v", f.Placements, f.PlacementsRead, listedAt) + } + listing = map[string]string{} + if msg, err = ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list); err != nil { + t.Fatal(err) + } + if f, _ = delivery.DecodeForge(msg.Data); !f.PlacementsRead || f.Placements != nil { + t.Errorf("a catalogue with no placement file says it read none: %v, %v", f.Placements, f.PlacementsRead) + } + listing = map[string]string{"gone.json": "b5"} + if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list); !errors.Is(err, ErrRetry) || !strings.Contains(err.Error(), "placement/gone.json") { + t.Errorf("a placement file the forge cannot give: %v", err) + } + failing := func(_, _, _, _ string) (map[string]string, error) { return nil, errors.New("forge down") } + if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, failing); !errors.Is(err, ErrRetry) { + t.Errorf("a listing the forge cannot give: %v", err) + } } func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2 diff --git a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go index 0b4258d..b0f0d8a 100644 --- a/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go +++ b/modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go @@ -91,6 +91,58 @@ func pinned(want string) *tls.Config { } } +// awaitedStore is a state bound when it appears (process.AwaitBucket) as the indexer's store: until then every call is +// process.ErrNotYet. +type awaitedStore struct{ b *process.AwaitedBucket } + +func (s awaitedStore) store() (kvStore, error) { + kv := s.b.KV() + if kv == nil { + return kvStore{}, process.ErrNotYet + } + return kvStore{kv}, nil +} + +func (s awaitedStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) { + st, err := s.store() + if err != nil { + return nil, 0, false, err + } + return st.Get(ctx, key) +} + +func (s awaitedStore) Create(ctx context.Context, key string, value []byte) error { + st, err := s.store() + if err != nil { + return err + } + return st.Create(ctx, key, value) +} + +func (s awaitedStore) Update(ctx context.Context, key string, value []byte, revision uint64) error { + st, err := s.store() + if err != nil { + return err + } + return st.Update(ctx, key, value, revision) +} + +func (s awaitedStore) Delete(ctx context.Context, key string, revision uint64) error { + st, err := s.store() + if err != nil { + return err + } + return st.Delete(ctx, key, revision) +} + +func (s awaitedStore) Keys(ctx context.Context) ([]string, error) { + st, err := s.store() + if err != nil { + return nil, err + } + return st.Keys(ctx) +} + // kvStore is the candidates bucket as the indexer's store: compare-and-set through Create, Update and Delete at a revision. type kvStore struct{ kv jetstream.KeyValue } @@ -253,6 +305,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro if err != nil { return err } + // Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a + // catalogue merge's placement is said and not written, and the rest of the indexer runs. + placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf) // The controller's build records only name the directory of a module whose key names none (fillPaths): without them // such a module waits for its path, every other is indexed (issue 1945). builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf) @@ -287,7 +342,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro } defer calls.Close() ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Builds: known, Logf: logger.Printf, - Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call)} + Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call), + Placement: awaitedStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }} // The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key // main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate. diff --git a/modules/mesh-delivery-indexer/go.mod b/modules/mesh-delivery-indexer/go.mod index 5a8d354..d272cd7 100644 --- a/modules/mesh-delivery-indexer/go.mod +++ b/modules/mesh-delivery-indexer/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-indexer go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-indexer/go.sum b/modules/mesh-delivery-indexer/go.sum index e793069..e8660dd 100644 --- a/modules/mesh-delivery-indexer/go.sum +++ b/modules/mesh-delivery-indexer/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-indexer/internal/index/candidates.go b/modules/mesh-delivery-indexer/internal/index/candidates.go index 592b309..cd48a70 100644 --- a/modules/mesh-delivery-indexer/internal/index/candidates.go +++ b/modules/mesh-delivery-indexer/internal/index/candidates.go @@ -60,6 +60,10 @@ type Indexer struct { // Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains). Contains func(ctx context.Context, repository, commit, in string) (bool, error) Logf func(format string, args ...any) + // Placement is the placement bucket (delivery.PlacementBucket, ADR 0549 §2); nil writes no placement. + Placement Store + // Say says one of the indexer's events (placement-changed); nil says none. + Say func(event string, body any) error mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach source sourceNow @@ -96,6 +100,9 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error { } ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945) } + if err := ix.applyPlacement(ctx, f); err != nil { + return err + } var digests map[string]string var held map[string]delivery.Module if f.Kind == delivery.ForgePull { diff --git a/modules/mesh-delivery-indexer/internal/index/manifests_test.go b/modules/mesh-delivery-indexer/internal/index/manifests_test.go index 08455bb..3cdd5f3 100644 --- a/modules/mesh-delivery-indexer/internal/index/manifests_test.go +++ b/modules/mesh-delivery-indexer/internal/index/manifests_test.go @@ -43,6 +43,8 @@ func withManifests(f delivery.Forge, manifests map[string]string, locks map[stri } func TestAMergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1 + // A merge's placement files are written as its manifests are (issue 1126): folded here to hold the count. + placementIsWrittenAtAMergeOnlyWhereItChanged(t) avahi := catalogueManifest(t, "avahi") ix, mods, manifests := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs)) f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "newtree"}), map[string]string{"modules/avahi": avahi}, nil) diff --git a/modules/mesh-delivery-indexer/internal/index/placement.go b/modules/mesh-delivery-indexer/internal/index/placement.go new file mode 100644 index 0000000..d611e68 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/placement.go @@ -0,0 +1,123 @@ +package index + +import ( + "context" + "errors" + "fmt" + "maps" + "slices" + + "git.novox.be/novox/mesh-sdk/go/delivery" + "git.novox.be/novox/mesh-sdk/go/process" +) + +// Placement of modules on nodes (novox/hq ADR 0549 §2, issue 1126): at each merge of the catalogue the forge bridge +// carries every placement file of the merge commit, and the indexer writes each node's key `placement.` in its +// state `placement` only when it differs (the revision alone is no change), deletes the key of a node whose file is +// gone, and says `placement-changed` for each node it wrote or deleted, naming the modules added, removed or changed. +// A merge that changes no placement writes nothing and says nothing. +// +// A file that does not read as a placement, or names another node than its file, is said in the log and its key is left +// as it was: the catalogue's own check refuses such a file before it merges (ADR 0549 §1), so it is a fault of that +// check, and a node keeps composing from the placement it had rather than from none. + +// applyPlacement takes the placement files of a merged message in; a message that read none (another repository, an +// older bridge) changes nothing. +func (ix *Indexer) applyPlacement(ctx context.Context, f delivery.Forge) error { + if ix.Placement == nil || f.Kind != delivery.ForgeMerged || !f.PlacementsRead { + return nil + } + keys, err := ix.Placement.Keys(ctx) + if errors.Is(err, process.ErrNotYet) { + // The state is not on the bus yet: said, and the merge's candidates are still taken out (issue 1790's binding + // writes the next catalogue merge's placement once it appears). + ix.logf("%s at %s: its placement files are not written: %v", f.Repository, f.Merge, err) + return nil + } + if err != nil { + return fmt.Errorf("reading the placement keys: %w", err) + } + gone := map[string]bool{} + for _, key := range keys { + if node, err := delivery.ParsePlacementKey(key); err == nil { + gone[node] = true + } + } + var refused []string + for _, node := range slices.Sorted(maps.Keys(f.Placements)) { + delete(gone, node) + next, err := delivery.DecodePlacement([]byte(f.Placements[node])) + if err == nil && next.Node != node { + err = fmt.Errorf("the file of %s names the node %q", node, next.Node) + } + if err != nil { + refused = append(refused, node) + ix.logf("%s at %s: the placement file of %s is refused and placement.%s is left as it was (the catalogue's check should have refused it): %v", + f.Repository, f.Merge, node, node, err) + continue + } + next.Revision = f.Merge + if err := ix.writePlacement(ctx, node, &next); err != nil { + return err + } + } + for _, node := range slices.Sorted(maps.Keys(gone)) { + if err := ix.writePlacement(ctx, node, nil); err != nil { + return err + } + } + if len(refused) > 0 { + // Said, and the merge's candidates are still taken out: a refused file holds nobody's delivery. + ix.logf("%s at %s: the placement files of %v do not read; their keys are left as they were", f.Repository, f.Merge, refused) + } + return nil +} + +// writePlacement writes one node's key, or deletes it for next nil, only when it differs, and says what changed. +func (ix *Indexer) writePlacement(ctx context.Context, node string, next *delivery.Placement) error { + key, err := delivery.PlacementKey(node) + if err != nil { + return err + } + var change *delivery.PlacementChange + err = write(ctx, ix.Placement, key, func(value []byte, found bool) ([]byte, bool, error) { + change = nil + var before delivery.Placement + if found { + cur, err := delivery.DecodePlacement(value) + if err != nil { + ix.logf("placement.%s holds what is not a placement, written again: %v", node, err) + } else { + before = cur + } + } + if next == nil { + if !found { + return nil, false, nil + } + c := delivery.ChangeOf(before, delivery.Placement{Node: node}) + change = &c + return nil, true, nil + } + if found && delivery.SamePlacement(before, *next) { + return nil, false, nil + } + data, err := delivery.EncodePlacement(*next) + if err != nil { + return nil, false, err + } + c := delivery.ChangeOf(before, *next) + change = &c + return data, false, nil + }) + if err != nil || change == nil { + return err + } + if ix.Say != nil { + if err := ix.Say(delivery.PlacementChangedEvent, *change); err != nil { + // The key is written: a reader that watches it hears the change; the event is said loudly as not said. + ix.logf("placement.%s is written, and %s could not be said: %v", node, delivery.PlacementChangedEvent, err) + } + } + return nil +} diff --git a/modules/mesh-delivery-indexer/internal/index/placement_test.go b/modules/mesh-delivery-indexer/internal/index/placement_test.go new file mode 100644 index 0000000..adb48f1 --- /dev/null +++ b/modules/mesh-delivery-indexer/internal/index/placement_test.go @@ -0,0 +1,97 @@ +package index + +import ( + "context" + "reflect" + "strings" + "testing" + + "git.novox.be/novox/mesh-sdk/go/delivery" +) + +// ADR 0549 §2 (issue 1126): a merge changing one node's file writes that key alone and says one event; a merge with no +// placement change writes nothing and says nothing; a node whose file is gone loses its key and says its modules +// removed; a file that does not read leaves its key; a message that read no placement changes none. +func placementIsWrittenAtAMergeOnlyWhereItChanged(t *testing.T) { + ctx := context.Background() + ix, _, logs := newIndexer(t) + place := newMem() + ix.Placement = place + var said []delivery.PlacementChange + ix.Say = func(event string, body any) error { + if event != delivery.PlacementChangedEvent { + t.Errorf("said %s", event) + } + said = append(said, body.(delivery.PlacementChange)) + return nil + } + merged := func(merge string, files map[string]string) delivery.Forge { + return delivery.Forge{V: delivery.Version, Kind: delivery.ForgeMerged, Repository: delivery.CatalogueRepository, Branch: "main", + Number: 7, Head: "h", Merge: merge, Root: "r", Trees: map[string]string{"": "t"}, Placements: files, PlacementsRead: true} + } + ace := `{"v":1,"node":"ace","modules":[{"module":"plex"}]}` + g14 := `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}` + written := func(node string) (delivery.Placement, uint64) { + t.Helper() + v, ok := place.vals["placement."+node] + if !ok { + return delivery.Placement{}, 0 + } + p, err := delivery.DecodePlacement(v) + if err != nil { + t.Fatal(err) + } + return p, place.revs["placement."+node] + } + + if err := ix.Apply(ctx, merged("m1", map[string]string{"ace": ace, "g14": g14})); err != nil { + t.Fatal(err) + } + if p, _ := written("ace"); p.Revision != "m1" || !reflect.DeepEqual(p.Assigned(), []string{"plex"}) || len(said) != 2 { + t.Fatalf("first merge: %+v, said %v", p, said) + } + _, aceRev := written("ace") + + said = nil + if err := ix.Apply(ctx, merged("m2", map[string]string{"ace": ace, "g14": g14})); err != nil { + t.Fatal(err) + } + if p, rev := written("ace"); rev != aceRev || p.Revision != "m1" || len(said) != 0 { + t.Errorf("a merge with no placement change wrote %+v (revision %d, was %d) and said %v", p, rev, aceRev, said) + } + + said = nil + if err := ix.Apply(ctx, merged("m3", map[string]string{"ace": ace, "g14": `{"v":1,"node":"g14","modules":[{"module":"zsh"},{"module":"xterm"}]}`})); err != nil { + t.Fatal(err) + } + if _, rev := written("ace"); rev != aceRev || len(said) != 1 || said[0].Node != "g14" || said[0].Revision != "m3" || + !reflect.DeepEqual(said[0].Added, []string{"xterm"}) { + t.Errorf("one file changed: ace's revision %d (was %d), said %+v", rev, aceRev, said) + } + + said = nil + if err := ix.Apply(ctx, merged("m4", map[string]string{"ace": ace, "g14": `{"v":1,"node":"ace","modules":[]}`})); err != nil { + t.Fatal(err) + } + if p, _ := written("g14"); len(p.Modules) != 2 || len(said) != 0 || !strings.Contains(logs.String(), "the placement file of %s is refused") { + t.Errorf("a file naming another node replaced the key: %+v, said %v, logs %s", p, said, logs) + } + + said = nil + if err := ix.Apply(ctx, merged("m5", map[string]string{"ace": ace})); err != nil { + t.Fatal(err) + } + if _, rev := written("g14"); rev != 0 || len(said) != 1 || !reflect.DeepEqual(said[0].Removed, []string{"xterm", "zsh"}) { + t.Errorf("a file removed: g14's key at revision %d, said %+v", rev, said) + } + + said = nil + unread := merged("m6", nil) + unread.PlacementsRead = false + if err := ix.Apply(ctx, unread); err != nil { + t.Fatal(err) + } + if p, _ := written("ace"); p.Revision != "m1" || len(said) != 0 { + t.Errorf("a message that read no placement changed it: %+v, %v", p, said) + } +} diff --git a/modules/mesh-delivery-indexer/internal/index/previews_test.go b/modules/mesh-delivery-indexer/internal/index/previews_test.go index 57e7b90..5c575a8 100644 --- a/modules/mesh-delivery-indexer/internal/index/previews_test.go +++ b/modules/mesh-delivery-indexer/internal/index/previews_test.go @@ -164,6 +164,7 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW } func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) { + t.Run("the same preview again", testTheSamePreviewAppliedAgainWritesNothing) ix, _, cands := previewBed(t, true) if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { t.Fatal(err) @@ -176,7 +177,7 @@ func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) } } -func TestTheSamePreviewAppliedAgainWritesNothing(t *testing.T) { +func testTheSamePreviewAppliedAgainWritesNothing(t *testing.T) { ix, _, cands := previewBed(t, true) if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { t.Fatal(err) diff --git a/modules/mesh-delivery-indexer/manifest_test.go b/modules/mesh-delivery-indexer/manifest_test.go index 46b3589..8becfb2 100644 --- a/modules/mesh-delivery-indexer/manifest_test.go +++ b/modules/mesh-delivery-indexer/manifest_test.go @@ -38,6 +38,7 @@ type manifest struct { Version string `json:"version"` Slug string `json:"slug"` Tools []string `json:"tools"` + Emits []string `json:"emits"` Consumes []string `json:"consumes"` State []state `json:"state"` Reads []string `json:"reads"` @@ -80,7 +81,11 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if m.Module != "mesh-delivery-indexer" { t.Errorf("module = %q", m.Module) } - if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { + // 1126 (ADR 0549 §2): the indexer's placement state, and the event it says when a node's placement changed. + if want := []string{"placement-changed"}; !reflect.DeepEqual(m.Emits, want) { + t.Errorf("emits = %v, want %v", m.Emits, want) + } + if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "placement", History: 10}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) { t.Errorf("state = %+v, want %+v", m.State, want) } if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) { @@ -107,8 +112,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build if want := []string{"indexer_take_in"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } - // No emits, no streams, no bus, no seat (ADR 0437 §4). - for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { + // No streams, no bus, no seat (ADR 0437 §4); its one event is placement-changed (ADR 0549 §2), checked above. + for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { if _, ok := top[k]; ok { t.Errorf("the manifest declares %q: the indexer has none", k) } diff --git a/modules/mesh-delivery-indexer/module.json b/modules/mesh-delivery-indexer/module.json index a30c24e..ce8432b 100644 --- a/modules/mesh-delivery-indexer/module.json +++ b/modules/mesh-delivery-indexer/module.json @@ -5,6 +5,9 @@ "tools": [ "indexer_take_in" ], + "emits": [ + "placement-changed" + ], "consumes": [ "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported", @@ -15,6 +18,7 @@ { "name": "candidates", "history": 10 }, { "name": "manifests", "history": 1 }, { "name": "modules", "history": 20 }, + { "name": "placement", "history": 10 }, { "name": "takeins", "history": 1 } ], "reads": [ diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go index c48e580..9f62582 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go @@ -50,6 +50,15 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool { mods, err := r.Modules(only) return map[string]any{"modules": mods}, err }}, + {Name: view.PlacementReadTool, + Description: "Where modules run, as the indexer read it from the catalogue's placement files at the last merge (placement., novox/hq ADR 0549): " + + "each node's modules in order with their host ports, bindings and settings placement, its pins and the seats its modules hold, and the catalogue revision. " + + "With node, that node's; with module, every node placing it; with neither, every node's. Keys that do not read are listed as unreadable. Read-only.", + Input: map[string]any{"type": "object", "properties": nodeModule}, + Run: func(args map[string]any) (any, error) { + node, module := nodeModuleOf(args) + return r.Placement(node, module) + }}, {Name: view.CandidateReadTool, Description: "What the indexer stored for one candidate (the open pull requests of a branch) under candidates., as stored: " + "the pull requests and heads, the base, and per pull request the touched modules with their new-form fingerprint (fp2) or why " + @@ -180,7 +189,8 @@ func read(state stdio.KeptState) view.StateRead { func main() { r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")), - WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have"))} + WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")), + PlacementState: read(stdio.State("mesh-delivery-indexer.placement"))} u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")), Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())} for _, lang := range delivery.AskLanguages { diff --git a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go index cd98d08..e71e4cf 100644 --- a/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go +++ b/modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go @@ -92,6 +92,32 @@ func TestModulesToolAnswersEveryKeyAsStored(t *testing.T) { if len(only) != 1 || string(only["b"]) != withNone { t.Errorf("none2 modules = %v", only) } + + // 1126 (ADR 0549 §2): indexer_placement answers a node's key as stored, every node placing a module, or every node's, + // and lists a key that does not read. + const ace, g14 = `{"v":1,"node":"ace","revision":"m1","modules":[{"module":"plex"}]}`, `{"v":1,"node":"g14","revision":"m1","modules":[{"module":"zsh"},{"module":"plex"}]}` + p := view.Reader{PlacementState: fake{"placement.ace": ace, "placement.g14": g14, "placement.bad": `{"v":1}`, "other": "{}"}.reader().State} + got := func(args map[string]any) map[string]json.RawMessage { + t.Helper() + out, err := run(t, p, view.PlacementReadTool, args) + if err != nil { + t.Fatal(err) + } + return out.(map[string]any)["placement"].(map[string]json.RawMessage) + } + if one := got(map[string]any{"node": "ace"}); len(one) != 1 || string(one["ace"]) != ace { + t.Errorf("node ace = %v", one) + } + if zsh := got(map[string]any{"module": "zsh"}); len(zsh) != 1 || string(zsh["g14"]) != g14 { + t.Errorf("module zsh = %v", zsh) + } + out, err = run(t, p, view.PlacementReadTool, map[string]any{}) + if all := out.(map[string]any); err != nil || len(all["placement"].(map[string]json.RawMessage)) != 2 || len(all["unreadable"].([]string)) != 2 { + t.Errorf("every node = %v, %v", all, err) + } + if _, err := run(t, p, view.PlacementReadTool, map[string]any{"node": "novox"}); err == nil || !strings.Contains(err.Error(), "no placement of novox") { + t.Errorf("a node with no key: %v", err) + } } const ( diff --git a/modules/mesh-delivery-view/go.mod b/modules/mesh-delivery-view/go.mod index 5222ff5..788bbc7 100644 --- a/modules/mesh-delivery-view/go.mod +++ b/modules/mesh-delivery-view/go.mod @@ -3,7 +3,7 @@ module mesh-delivery-view go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026202658-805b4d9a3536 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 github.com/nats-io/nats-server/v2 v2.11.17 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-delivery-view/go.sum b/modules/mesh-delivery-view/go.sum index c6d0c7b..e8660dd 100644 --- a/modules/mesh-delivery-view/go.sum +++ b/modules/mesh-delivery-view/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026202658-805b4d9a3536 h1:+SoDdDXF7Tgj38WP4kRpUSps6DJVl6P+w5u/KfdtwnQ= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026202658-805b4d9a3536/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-delivery-view/internal/view/read.go b/modules/mesh-delivery-view/internal/view/read.go index 31be53d..4ca99eb 100644 --- a/modules/mesh-delivery-view/internal/view/read.go +++ b/modules/mesh-delivery-view/internal/view/read.go @@ -20,6 +20,8 @@ import ( const ( ModuleReadTool = "indexer_module" ModulesReadTool = "indexer_modules" + // PlacementReadTool reads the indexer's placement (novox/hq ADR 0549 §2, issue 1126). + PlacementReadTool = "indexer_placement" ) // StateRead reads the indexer's modules state, as the view's reads grant it: its keys and a key's stored value (nil for a key with none). @@ -39,6 +41,7 @@ type Reader struct { WantsState StateRead // the assigner's wants and holds DifferencesState StateRead // the assigner's differences HavesState StateRead // the node-engine's haves + PlacementState StateRead // the indexer's placement } // Module is the stored value of modules., exactly as the indexer wrote it. @@ -613,3 +616,55 @@ func (r Reader) Differences(node, module string) (KeptAnswer, error) { func (r Reader) Haves(node, module string) (KeptAnswer, error) { return r.kept(r.HavesState, "the haves", node, module) } + +// Placement is the indexer's placement as stored (ADR 0549 §2): one node's key with node, every node placing a module +// with module (each node's key whole, so its pins, ports and bindings are beside the module), and every node's key with +// neither. A key that does not read as a placement is listed as unreadable, never dropped. +func (r Reader) Placement(node, module string) (map[string]any, error) { + node, module = strings.TrimSpace(node), strings.TrimSpace(module) + if r.PlacementState.Keys == nil || r.PlacementState.Get == nil { + return nil, fmt.Errorf("the placement state cannot be read here") + } + if node != "" { + key, err := delivery.PlacementKey(node) + if err != nil { + return nil, err + } + raw, err := r.PlacementState.Get(key) + if err != nil { + return nil, err + } + if len(raw) == 0 { + return nil, fmt.Errorf("no placement of %s: the catalogue holds no placement file for it, or the indexer has read none yet", node) + } + return map[string]any{"placement": map[string]json.RawMessage{node: raw}}, nil + } + keys, err := r.PlacementState.Keys() + if err != nil { + return nil, err + } + sort.Strings(keys) + values, err := r.PlacementState.values(keys) + if err != nil { + return nil, err + } + out := map[string]json.RawMessage{} + var unreadable []string + for i, key := range keys { + n, err := delivery.ParsePlacementKey(key) + if err != nil { + unreadable = append(unreadable, key) + continue + } + p, err := delivery.DecodePlacement(values[i]) + if err != nil { + unreadable = append(unreadable, key) + continue + } + if module != "" && !slices.Contains(p.Assigned(), module) { + continue + } + out[n] = values[i] + } + return map[string]any{"placement": out, "unreadable": unreadable}, nil +} diff --git a/modules/mesh-delivery-view/manifest_test.go b/modules/mesh-delivery-view/manifest_test.go index 6d41a31..b887e2b 100644 --- a/modules/mesh-delivery-view/manifest_test.go +++ b/modules/mesh-delivery-view/manifest_test.go @@ -84,7 +84,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 if m.Module != "mesh-delivery-view" { t.Errorf("module = %q", m.Module) } - want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 + want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9 got := slices.Clone(m.Reads) slices.Sort(got) if !reflect.DeepEqual(got, want) { @@ -115,7 +115,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7 } // The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer // itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads. - if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage"}; !reflect.DeepEqual(m.Tools, want) { + if want := []string{"indexer_module", "indexer_modules", "indexer_placement", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage"}; !reflect.DeepEqual(m.Tools, want) { t.Errorf("tools = %v, want %v", m.Tools, want) } for _, a := range m.Build.Artifacts { diff --git a/modules/mesh-delivery-view/module.json b/modules/mesh-delivery-view/module.json index 9e10b47..a99a8f6 100644 --- a/modules/mesh-delivery-view/module.json +++ b/modules/mesh-delivery-view/module.json @@ -6,6 +6,7 @@ "tools": [ "indexer_module", "indexer_modules", + "indexer_placement", "indexer_candidate", "indexer_candidates", "builder_build", @@ -45,6 +46,7 @@ "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", diff --git a/modules/mesh-grants/go.mod b/modules/mesh-grants/go.mod index 6d8f4e9..19d6941 100644 --- a/modules/mesh-grants/go.mod +++ b/modules/mesh-grants/go.mod @@ -3,7 +3,7 @@ module mesh-grants go 1.26.0 require ( - git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 + git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 github.com/nats-io/nats.go v1.54.0 ) diff --git a/modules/mesh-grants/go.sum b/modules/mesh-grants/go.sum index 4cc1a88..8035b76 100644 --- a/modules/mesh-grants/go.sum +++ b/modules/mesh-grants/go.sum @@ -1,5 +1,5 @@ -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583 h1:ikD/ggf9/5k23sxJOX1PFcGzb4M3+QeC66AT0MyQ9Zs= -git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk= +git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0= github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo= diff --git a/modules/mesh-grants/grants_test.go b/modules/mesh-grants/grants_test.go index 77842e3..3e72e77 100644 --- a/modules/mesh-grants/grants_test.go +++ b/modules/mesh-grants/grants_test.go @@ -708,6 +708,43 @@ func TestGrants(t *testing.T) { } }) + // 1126 (ADR 0549 §1, §3): once the catalogue places the nodes, a seat held once for the mesh is held where placement + // says, not on every machine running a module that claims it; what placement adds is composed beside the assignments. + t.Run("a mesh seat is held where placement says", func(t *testing.T) { + store := catalogue.Manifest{Module: "store", Version: "1", + DefinesSeats: []catalogue.SeatDeclaration{{Name: "store-seat", Scope: catalogue.ScopeMesh, Serves: []catalogue.Verb{{Name: "databases"}}}}, + Claims: []catalogue.Claim{{Name: "store-seat", Scope: catalogue.ScopeMesh}}} + in := grants.Inputs{Nodes: []string{"a", "b"}, Assigned: map[string][]string{"a": {"store"}, "b": {"store"}}, + Manifests: map[string]catalogue.Manifest{"store": store}} + holds := func(recs buscompose.Records, node string) int { + n := 0 + for _, d := range recs.Assigned[node] { + n += len(d.Holds) + } + return n + } + if recs, _ := grants.Records(in); holds(recs, "a") != 1 || holds(recs, "b") != 1 { + t.Fatalf("before placement the seat is held on a %d and b %d times, want both", holds(recs, "a"), holds(recs, "b")) + } + in.Placed = map[string]delivery.Placement{ + "a": {Node: "a", Modules: []delivery.PlacedModule{{Module: "store"}}, Holds: []delivery.Holding{{Claim: "store-seat", Scope: "mesh", Module: "store"}}}, + "b": {Node: "b", Modules: []delivery.PlacedModule{{Module: "store"}}}} + if recs, _ := grants.Records(in); holds(recs, "a") != 1 || holds(recs, "b") != 0 { + t.Errorf("placed, the seat is held on a %d and b %d times, want a alone", holds(recs, "a"), holds(recs, "b")) + } + b := fixture() + pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "b", Modules: []delivery.PlacedModule{{Module: "mod-y"}, {Module: "mod-x"}}}) + b.states[grants.ReadPlacement] = map[string][]byte{"placement.b": pl} + read := grants.ReadInputs(b, &grants.Manifests{}) + if !slices.Contains(read.Assigned["b"], "mod-x") || read.Placed["b"].Node != "b" || slices.ContainsFunc(read.Gaps, func(g grants.Gap) bool { return g.Input == "seat holdings" }) { + t.Errorf("read: assigned %v, placed %v, gaps %v", read.Assigned, read.Placed, read.Gaps) + } + b.refused[grants.ReadPlacement] = true + if read := grants.ReadInputs(b, &grants.Manifests{}); slices.ContainsFunc(read.Gaps, func(g grants.Gap) bool { return g.Input == grants.ReadPlacement && g.Failed }) { + t.Errorf("placement not on the bus is a failed read: %v", read.Gaps) + } + }) + t.Run("compare excuses a retired verb and a multiply claimed mesh seat's added subscriptions on its machines, nothing else", func(t *testing.T) { const retired, store, storeInfo = "mesh.seat.mesh-delivery.tool.mark-ready", "mesh.seat.mesh-store.tool.databases", "$SRV.INFO.mesh-store.>" two := map[string][]string{"mesh-store": {"a", "n"}} diff --git a/modules/mesh-grants/internal/grants/follow.go b/modules/mesh-grants/internal/grants/follow.go index 938a243..3d19217 100644 --- a/modules/mesh-grants/internal/grants/follow.go +++ b/modules/mesh-grants/internal/grants/follow.go @@ -18,7 +18,7 @@ type Watcher interface { // Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved // config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's // wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts. -var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts} +var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement} // Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot // open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect). diff --git a/modules/mesh-grants/internal/grants/inputs.go b/modules/mesh-grants/internal/grants/inputs.go index fdb6c03..0634fe5 100644 --- a/modules/mesh-grants/internal/grants/inputs.go +++ b/modules/mesh-grants/internal/grants/inputs.go @@ -34,11 +34,14 @@ const ( ReadBuilds = "mesh-delivery-builder.builds" // ReadAccounts is the vault's bcrypt hashes of the bus accounts it holds, by bus user (ADR 0547 clause 9). ReadAccounts = "mesh-vault.bus-accounts" + // ReadPlacement is the indexer's placement, read from the catalogue (ADR 0549 §2, §3): each node's modules and the seats + // they hold. + ReadPlacement = "mesh-delivery-indexer.placement" ) // Reads is every state the part reads, in the manifest's order. var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds, - ReadAccounts} + ReadAccounts, ReadPlacement} // The bus module's tools the part calls, as its manifest's invokes name them: read-only, each. const ( @@ -103,8 +106,12 @@ type resolved struct { type Inputs struct { // Nodes is every machine seen: in a node agent's have or a standing assignment. Nodes []string - // Assigned is each machine's standing assignments, sorted. + // Assigned is each machine's standing assignments and what the catalogue's placement puts there, sorted: while both + // stand (ADR 0549 §5) a module either names keeps its membership. Assigned map[string][]string + // Placed is each node's placement as the indexer read it from the catalogue (ADR 0549 §2); empty before any is read. + // Where it is, a seat held once for the mesh is held where placement says, not wherever a module claiming it runs. + Placed map[string]delivery.Placement // Manifests is each module's manifest at the indexer's trunk, by module. Manifests map[string]catalogue.Manifest // Digests is each module's manifest digest, by module. @@ -230,6 +237,38 @@ func ReadInputs(r Reader, cache *Manifests) Inputs { in.BusNode = a.Node } } + // The catalogue's placement (ADR 0549 §3): what it places is composed beside the assignments, and the seats held once + // for the mesh are held where it says. Not on the bus yet is a standing gap, never a failed read, so the list is still + // composed and sealed from the assignments alone. + if keys, err := r.Keys(ReadPlacement); err != nil { + in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement, Why: "not read yet: the assignments alone place modules (" + err.Error() + ")"}) + } else { + for _, k := range keys { + raw, found, err := r.Get(ReadPlacement, k) + if err != nil || !found { + continue + } + p, err := delivery.DecodePlacement(raw) + node, kerr := delivery.ParsePlacementKey(k) + if err != nil || kerr != nil || p.Node != node { + in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement + " " + k, Why: "not a placement, so the node is composed from its assignments alone"}) + continue + } + if in.Placed == nil { + in.Placed = map[string]delivery.Placement{} + } + in.Placed[node] = p + nodes[node] = true + for _, m := range p.Assigned() { + if !slices.Contains(in.Assigned[node], m) { + in.Assigned[node] = append(in.Assigned[node], m) + } + } + } + } + if len(in.Placed) > 0 { + in.Gaps = slices.DeleteFunc(in.Gaps, func(g Gap) bool { return g.Input == "seat holdings" }) + } haveKeys, err := r.Keys(ReadHave) if err != nil { in.Gaps = append(in.Gaps, Gap{Input: ReadHave, Why: "refused or absent: " + err.Error(), Failed: true}) diff --git a/modules/mesh-grants/internal/grants/records.go b/modules/mesh-grants/internal/grants/records.go index e53beab..e92e11c 100644 --- a/modules/mesh-grants/internal/grants/records.go +++ b/modules/mesh-grants/internal/grants/records.go @@ -7,6 +7,7 @@ import ( "git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/catalogue" + "git.novox.be/novox/mesh-sdk/go/delivery" ) // Records is what the bus composer composes from, rebuilt from the bus (mesh-controller internal/inventory busrecords.go, @@ -59,6 +60,9 @@ func Records(in Inputs) (buscompose.Records, []Gap) { continue } d := declaredFor(m, seats, declarers) + if len(in.Placed) > 0 { + d.Holds = heldWherePlaced(d.Holds, in.Placed[node], m.Module) + } for _, sr := range catalogue.ReadsGranted(m, here, node) { d.KeyedReads = append(d.KeyedReads, buscompose.KeyedRead{Read: sr.Read, Key: sr.Key}) } @@ -299,3 +303,22 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node } return out, touches, onEvents } + +// heldWherePlaced is a module's holds on a node as the catalogue's placement says them (ADR 0549 §1, §3): a seat held once +// for the mesh is held there only when the node's placement names this module holding it, so a module claiming such a seat +// on two machines holds it on the one placement names, not on both; a seat of every node is held wherever the module runs. +func heldWherePlaced(holds []buscompose.Seat, p delivery.Placement, module string) []buscompose.Seat { + var out []buscompose.Seat + for _, h := range holds { + if h.Scope != "" && h.Scope != catalogue.ScopeMesh { + out = append(out, h) + continue + } + if slices.ContainsFunc(p.Holds, func(x delivery.Holding) bool { + return x.Claim == h.Name && x.Module == module && x.Kind == h.Kind + }) { + out = append(out, h) + } + } + return out +} diff --git a/modules/mesh-grants/module.json b/modules/mesh-grants/module.json index 710c0ce..65d06b0 100644 --- a/modules/mesh-grants/module.json +++ b/modules/mesh-grants/module.json @@ -10,6 +10,7 @@ "mesh-delivery-builder.builds", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules", + "mesh-delivery-indexer.placement", "node-engine.have", "mesh-vault.bus-accounts" ],