Merge pull request 'Placement from the catalogue: the bridge carries it, the indexer writes it, the assigner compares it (issue 1126)' (#76) from feat/1126-nodes-compose into main

This commit was merged in pull request #76.
This commit is contained in:
2026-10-26 21:11:26 +00:00
46 changed files with 1246 additions and 59 deletions
+3
View File
@@ -1 +1,4 @@
vendor/ 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
@@ -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. // 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket.
func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) { func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
for _, missing := range readBuckets { for _, missing := range readBuckets {
if missing == delivery.PlacementBucket {
continue // it starts without it, below
}
b := startBed(t, missing) b := startBed(t, missing)
select { select {
case err := <-b.start(t): 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") 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. // putJSON writes a value into one of the bed's buckets.
@@ -406,6 +437,32 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
if !cleared { if !cleared {
t.Errorf("the condition of web on neither list on g14 was not cleared once g14 got its want") 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 // 1297/B9 (the assigner's half, ADR 0528 clause 6): started whole, the assigner holds a fingerprint whose health check
@@ -59,6 +59,7 @@ var readBuckets = []string{
delivery.CandidatesBucket, delivery.CandidatesBucket,
delivery.ManifestsBucket, delivery.ManifestsBucket,
delivery.ModulesBucket, delivery.ModulesBucket,
delivery.PlacementBucket,
delivery.PresenceBucket, delivery.PresenceBucket,
delivery.HaveBucket, delivery.HaveBucket,
delivery.StepsBucket(goBuildModule), 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) logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err)
continue 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 { if err != nil {
return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err) 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("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)) 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. // 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.BuilderBuildsBucket,
delivery.CandidatesBucket, delivery.CandidatesBucket,
delivery.ModulesBucket, delivery.ModulesBucket,
delivery.PlacementBucket,
delivery.PresenceBucket, delivery.PresenceBucket,
delivery.HaveBucket, delivery.HaveBucket,
delivery.AssignerAsksBucket, 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 // 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. // 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() w := newWorld()
dirty := make(chan struct{}, 1) dirty := make(chan struct{}, 1)
wake := func() { 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) logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket)
} }
started := 0
for _, name := range watched { 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 var opts []jetstream.WatchOpt
if name == delivery.HaveBucket { if name == delivery.HaveBucket {
opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back 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 { if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil {
return err 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 { select {
case <-ctx.Done(): case <-ctx.Done():
return nil return nil
@@ -297,6 +323,10 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
case err := <-errs: case err := <-errs:
return err return err
case <-settled: case <-settled:
case kv := <-placementAppeared:
if err := takePlacement(kv); err != nil {
return err
}
case <-dirty: case <-dirty:
reopenPuts(p) reopenPuts(p)
p.pass(ctx) p.pass(ctx)
@@ -94,7 +94,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}} kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}}
kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true
kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote 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 { select {
case key := <-wrote: case key := <-wrote:
if key != "want.ace.node" { if key != "want.ace.node" {
@@ -60,7 +60,23 @@ func (p *passer) pass(ctx context.Context) {
// controller has composed it and again whenever that digest changes (ADR 0539 §6). // controller has composed it and again whenever that digest changes (ADR 0539 §6).
func (p *passer) nodeWants(ctx context.Context) { func (p *passer) nodeWants(ctx context.Context) {
written := p.w.read(delivery.AssignerWantsBucket) 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() { for node, d := range p.w.nodeComposed() {
if placed[node] {
continue
}
key, err := delivery.WantNodeKey(node) key, err := delivery.WantNodeKey(node)
if err != nil { if err != nil {
continue continue
@@ -97,6 +113,8 @@ func (p *passer) wants(ctx context.Context) {
} }
} }
wanted := p.w.wantsOf() wanted := p.w.wantsOf()
frozen := p.sayPlacement(p.w.placementDiffers())
placed := p.w.placedNodes()
puts := p.putsByModule() puts := p.putsByModule()
p.staleAsks(ctx, trunk) p.staleAsks(ctx, trunk)
names := map[string]bool{} names := map[string]bool{}
@@ -119,7 +137,14 @@ func (p *passer) wants(ctx context.Context) {
fp := trunk[name].FP2 fp := trunk[name].FP2
name := name name := name
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name], Now: time.Now(), Puts: puts[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 { GrantedAt: func(fp string) bool {
key, err := delivery.ControllerGrantKey(name, fp) key, err := delivery.ControllerGrantKey(name, fp)
return err == nil && p.w.has(delivery.ControllerGrantsBucket, key) 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) last[node] = p.w.lastHealthy(node, name, fp)
} }
out := assign.Wants(m, nodes, wanted[name], last) out := assign.Wants(m, nodes, wanted[name], last)
p.yeses(ctx, name, m, placed, wanted[name], &out)
gatherNeither(neither, name, out) gatherNeither(neither, name, out)
leftNeither(left, m, wanted[name], out) leftNeither(left, m, wanted[name], out)
p.sayNoWant(name, m.Nodes, 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) p.recordAsk(ctx, name, fp, need)
} }
for node, want := range out.Wants { 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) key, err := delivery.WantKey(node, name)
if err != nil { if err != nil {
p.logf("no want of %s on %s is written: %v", name, node, err) 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 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
}
}
}
@@ -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 // 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. // the journal, so it is loud with no operator channel.
func TestAModuleOnNeitherListIsRaisedOnceAndClearedOnce(t *testing.T) { 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)) p := newPasser(buckets{}, time.Date(2026, 10, 25, 15, 0, 0, 0, time.UTC))
var said []asks.Condition var said []asks.Condition
failing := true 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. // 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{} got := map[string]neitherOne{}
gatherNeither(got, "screen-lock", assign.Outcome{Neither: map[string]assign.Holding{"g14": {Rule: delivery.RuleGrant, Words: "w"}}}) gatherNeither(got, "screen-lock", assign.Outcome{Neither: map[string]assign.Holding{"g14": {Rule: delivery.RuleGrant, Words: "w"}}})
gatherNeither(got, "rofi", assign.Outcome{}) 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 // 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. // every module that is not held. Clearing a condition that is not open is quiet at the router.
func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) { func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) {
t.Run("a held module", testAHeldModuleRaisesNoNeitherAndClearsAStandingOne)
wantsKV := &memKV{values: map[string][]byte{}} wantsKV := &memKV{values: map[string][]byte{}}
p := newPasser(buckets{delivery.AssignerWantsBucket: wantsKV, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now()) p := newPasser(buckets{delivery.AssignerWantsBucket: wantsKV, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now())
var said []asks.Condition 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 // 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 // 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. // raised (the first pass after a start), each once.
func TestAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) { func testAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) {
setup := func() (*passer, *[]asks.Condition) { setup := func() (*passer, *[]asks.Condition) {
p := newPasser(buckets{delivery.AssignerWantsBucket: &memKV{values: map[string][]byte{}}, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now()) p := newPasser(buckets{delivery.AssignerWantsBucket: &memKV{values: map[string][]byte{}}, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now())
said := &[]asks.Condition{} 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 // (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. // yes the want names the revised digest.
func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing.T) { 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) fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64)
revised, was := "sha256:"+strings.Repeat("e", 64), "sha256:"+strings.Repeat("d", 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{}} 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") 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)
}
}
@@ -14,6 +14,8 @@ import (
"github.com/nats-io/nats.go/jetstream" "github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/catalogue" "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/delivery"
"git.novox.be/novox/mesh-sdk/go/provider" "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 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) { 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{} out, all := map[string][]string{}, map[string]assignment{}
for _, value := range w.read(provider.AssignmentsBucket) { for _, value := range w.read(provider.AssignmentsBucket) {
var a assignment var a assignment
@@ -210,9 +260,86 @@ func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh {
} }
m.Manifests[name] = p.manifest 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 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. // 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) { 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{}, ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{},
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
go 1.26.0 go 1.26.0
require ( 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-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -10
View File
@@ -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.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
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.20261026210326-979644420117/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=
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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -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 // 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). // composer at another mesh-sdk reads (issue 1622).
Digests map[string]string 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 // Change is a group's change as the indexer's candidate holds it: the manifest of every module it moves or adds, those
@@ -42,8 +42,15 @@ func HeadJob(m Mesh, c Change) (delivery.ComposeJob, error) {
for _, name := range c.Removed { for _, name := range c.Removed {
delete(change, name) delete(change, name)
} }
req := composer.Request{V: composer.Version, Input: plainInput(m.Assigned), Depth: composer.DepthResolve, in, fromClaims := plainInput(m.Assigned), true
HoldingsFromClaims: true, Machines: Touched(m, c), 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}}} 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) 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) readerFails, readerWarnings := judgeReaders(c, change, readers)
fails = append(fails, readerFails...) fails = append(fails, readerFails...)
v.Warnings = append(v.Warnings, readerWarnings...) 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) return finish(v, m, c, machines, fails)
} }
@@ -8,6 +8,7 @@ import (
"time" "time"
"git.novox.be/novox/mesh-sdk/go/catalogue" "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/composer"
"git.novox.be/novox/mesh-sdk/go/delivery" "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 { 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) 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 // 1622/AC2, AC3: a catalogue member using a field only the SDK head reads passes with the head's composer, and the
@@ -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)}
}
@@ -34,7 +34,7 @@ func TestManifest(t *testing.T) {
slices.Sort(reads) slices.Sort(reads)
want := []string{"mesh-build-go.steps", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", 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-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) { if !reflect.DeepEqual(reads, want) {
t.Errorf("reads = %v, want exactly %v", reads, want) t.Errorf("reads = %v, want exactly %v", reads, want)
} }
@@ -12,6 +12,7 @@
"mesh-delivery-indexer.candidates", "mesh-delivery-indexer.candidates",
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules", "mesh-delivery-indexer.modules",
"mesh-delivery-indexer.placement",
"mesh-delivery-teller.answers", "mesh-delivery-teller.answers",
"mesh-delivery-watchman.presence", "mesh-delivery-watchman.presence",
"node-engine.have" "node-engine.have"
@@ -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): // 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 // 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 <path>: 404 <body>` (client.ts request). // (client.ts getFile). A path the forge does not have fails with `Gitea API <path>: 404 <body>` (client.ts request).
@@ -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 // 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 // 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). // 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) { func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, logger *log.Logger) {
msg, err := bridge.Convert(m.Subject(), m.Data(), trees, files) msg, err := bridge.ConvertWith(m.Subject(), m.Data(), trees, files, list)
switch { switch {
case errors.Is(err, bridge.ErrRetry): case errors.Is(err, bridge.ErrRetry):
logger.Printf("%v; heard again in %s", err, retryDelay) 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. // Bounded: the turn is held, and no sync or other event goes on without it.
hctx, cancel := context.WithTimeout(ctx, syncDeadline) hctx, cancel := context.WithTimeout(ctx, syncDeadline)
defer cancel() 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) })) jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
@@ -81,7 +81,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit
}) })
var logs bytes.Buffer var logs bytes.Buffer
m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{} 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" { 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) 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 func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5
var logs bytes.Buffer var logs bytes.Buffer
m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(map[string]any{"head_sha": "short"})}, &fakePub{} 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 { if len(pub.subjects) != 0 {
t.Errorf("a message was written: %v", pub.subjects) 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) func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here)
var logs bytes.Buffer var logs bytes.Buffer
m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")} 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 { if m.acked != 0 || m.termd != 0 {
t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd) 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) { call := func(context.Context, string, any) (json.RawMessage, error) {
return nil, errors.New("the tree listing of novox/mesh-delivery was cut") 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 { if len(pub.subjects) != 0 {
t.Errorf("a message was written: %v", pub.subjects) 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 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 { 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) t.Fatalf("calls %v, want the trees once and first", keys)
} }
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge
go 1.26.0 go 1.26.0
require ( 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-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -4
View File
@@ -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.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
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.20261026210326-979644420117/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=
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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -7,6 +7,7 @@ import (
"encoding/json" "encoding/json"
"errors" "errors"
"fmt" "fmt"
"maps"
"path" "path"
"regexp" "regexp"
"slices" "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 // 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. // 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) { 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 var action string
switch subject { switch subject {
case UpdatedSubject: case UpdatedSubject:
@@ -114,7 +126,7 @@ func Convert(subject string, data []byte, trees Trees, files Files) (Message, er
return Message{}, unreadable(subject, "no branch") return Message{}, unreadable(subject, "no branch")
} }
if subject == MergedSubject && p.Base == Trunk { 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, f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number,
Head: p.HeadSHA, Action: action} 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. // 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) { if !sha40.MatchString(p.MergeSHA) {
return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", 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 { if err != nil {
return Message{}, unreadable(subject, "%v", err) return Message{}, unreadable(subject, "%v", err)
} }
body, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number, 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}) 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 { if err != nil {
return Message{}, unreadable(subject, "%v", err) return Message{}, unreadable(subject, "%v", err)
} }
@@ -350,3 +369,30 @@ func OpenMessage(answer []byte) (Message, error) {
} }
return Message{Event: delivery.OpenEvent(), Data: body}, nil 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 `<node>.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
}
@@ -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) { 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) 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 func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2
@@ -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. // 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 } 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 { if err != nil {
return err 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 // 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). // 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) 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() defer calls.Close()
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Builds: known, Logf: logger.Printf, 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 // 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 // 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. // not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate.
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-indexer
go 1.26.0 go 1.26.0
require ( 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-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -2
View File
@@ -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.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
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/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -60,6 +60,10 @@ type Indexer struct {
// Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains). // 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) Contains func(ctx context.Context, repository, commit, in string) (bool, error)
Logf func(format string, args ...any) 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 mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
source sourceNow 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) 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 digests map[string]string
var held map[string]delivery.Module var held map[string]delivery.Module
if f.Kind == delivery.ForgePull { if f.Kind == delivery.ForgePull {
@@ -43,6 +43,8 @@ func withManifests(f delivery.Forge, manifests map[string]string, locks map[stri
} }
func TestAMergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1 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") avahi := catalogueManifest(t, "avahi")
ix, mods, manifests := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs)) 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) f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "newtree"}), map[string]string{"modules/avahi": avahi}, nil)
@@ -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.<node>` 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
}
@@ -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)
}
}
@@ -164,6 +164,7 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW
} }
func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) { func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) {
t.Run("the same preview again", testTheSamePreviewAppliedAgainWritesNothing)
ix, _, cands := previewBed(t, true) ix, _, cands := previewBed(t, true)
if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil {
t.Fatal(err) 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) ix, _, cands := previewBed(t, true)
if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil { if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil {
t.Fatal(err) t.Fatal(err)
@@ -38,6 +38,7 @@ type manifest struct {
Version string `json:"version"` Version string `json:"version"`
Slug string `json:"slug"` Slug string `json:"slug"`
Tools []string `json:"tools"` Tools []string `json:"tools"`
Emits []string `json:"emits"`
Consumes []string `json:"consumes"` Consumes []string `json:"consumes"`
State []state `json:"state"` State []state `json:"state"`
Reads []string `json:"reads"` 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" { if m.Module != "mesh-delivery-indexer" {
t.Errorf("module = %q", m.Module) 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) 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) { 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) { if want := []string{"indexer_take_in"}; !reflect.DeepEqual(m.Tools, want) {
t.Errorf("tools = %v, want %v", m.Tools, want) t.Errorf("tools = %v, want %v", m.Tools, want)
} }
// No emits, no streams, no bus, no seat (ADR 0437 §4). // 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", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} { for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
if _, ok := top[k]; ok { if _, ok := top[k]; ok {
t.Errorf("the manifest declares %q: the indexer has none", k) t.Errorf("the manifest declares %q: the indexer has none", k)
} }
@@ -5,6 +5,9 @@
"tools": [ "tools": [
"indexer_take_in" "indexer_take_in"
], ],
"emits": [
"placement-changed"
],
"consumes": [ "consumes": [
"mesh-build-go.toolchain.reported", "mesh-build-go.toolchain.reported",
"mesh-build-node.toolchain.reported", "mesh-build-node.toolchain.reported",
@@ -15,6 +18,7 @@
{ "name": "candidates", "history": 10 }, { "name": "candidates", "history": 10 },
{ "name": "manifests", "history": 1 }, { "name": "manifests", "history": 1 },
{ "name": "modules", "history": 20 }, { "name": "modules", "history": 20 },
{ "name": "placement", "history": 10 },
{ "name": "takeins", "history": 1 } { "name": "takeins", "history": 1 }
], ],
"reads": [ "reads": [
@@ -50,6 +50,15 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool {
mods, err := r.Modules(only) mods, err := r.Modules(only)
return map[string]any{"modules": mods}, err 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.<node>, 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, {Name: view.CandidateReadTool,
Description: "What the indexer stored for one candidate (the open pull requests of a branch) under candidates.<group>, as stored: " + Description: "What the indexer stored for one candidate (the open pull requests of a branch) under candidates.<group>, as stored: " +
"the pull requests and heads, the base, and per pull request the touched modules with their new-form fingerprint (fp2) or why " + "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() { 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")), 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")), 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())} Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())}
for _, lang := range delivery.AskLanguages { for _, lang := range delivery.AskLanguages {
@@ -92,6 +92,32 @@ func TestModulesToolAnswersEveryKeyAsStored(t *testing.T) {
if len(only) != 1 || string(only["b"]) != withNone { if len(only) != 1 || string(only["b"]) != withNone {
t.Errorf("none2 modules = %v", only) 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 ( const (
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-view
go 1.26.0 go 1.26.0
require ( 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-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0 github.com/nats-io/nats.go v1.54.0
) )
+2 -2
View File
@@ -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.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
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/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
@@ -20,6 +20,8 @@ import (
const ( const (
ModuleReadTool = "indexer_module" ModuleReadTool = "indexer_module"
ModulesReadTool = "indexer_modules" 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). // 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 WantsState StateRead // the assigner's wants and holds
DifferencesState StateRead // the assigner's differences DifferencesState StateRead // the assigner's differences
HavesState StateRead // the node-engine's haves HavesState StateRead // the node-engine's haves
PlacementState StateRead // the indexer's placement
} }
// Module is the stored value of modules.<module>, exactly as the indexer wrote it. // Module is the stored value of modules.<module>, 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) { func (r Reader) Haves(node, module string) (KeptAnswer, error) {
return r.kept(r.HavesState, "the haves", node, module) 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
}
+2 -2
View File
@@ -84,7 +84,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
if m.Module != "mesh-delivery-view" { if m.Module != "mesh-delivery-view" {
t.Errorf("module = %q", m.Module) 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) got := slices.Clone(m.Reads)
slices.Sort(got) slices.Sort(got)
if !reflect.DeepEqual(got, want) { 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 // 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. // 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) t.Errorf("tools = %v, want %v", m.Tools, want)
} }
for _, a := range m.Build.Artifacts { for _, a := range m.Build.Artifacts {
+2
View File
@@ -6,6 +6,7 @@
"tools": [ "tools": [
"indexer_module", "indexer_module",
"indexer_modules", "indexer_modules",
"indexer_placement",
"indexer_candidate", "indexer_candidate",
"indexer_candidates", "indexer_candidates",
"builder_build", "builder_build",
@@ -45,6 +46,7 @@
"mesh-delivery-gate.source", "mesh-delivery-gate.source",
"mesh-delivery-indexer.candidates", "mesh-delivery-indexer.candidates",
"mesh-delivery-indexer.modules", "mesh-delivery-indexer.modules",
"mesh-delivery-indexer.placement",
"mesh-delivery-teller.answers", "mesh-delivery-teller.answers",
"mesh-delivery-watchman.presence", "mesh-delivery-watchman.presence",
"mesh-delivery-watchman.stuck", "mesh-delivery-watchman.stuck",
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-grants
go 1.26.0 go 1.26.0
require ( 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 github.com/nats-io/nats.go v1.54.0
) )
+2 -2
View File
@@ -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.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
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/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 h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI= 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= github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
+37
View File
@@ -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) { 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.>" 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"}} two := map[string][]string{"mesh-store": {"a", "n"}}
@@ -18,7 +18,7 @@ type Watcher interface {
// Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved // 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 // 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. // 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 // 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). // open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).
+41 -2
View File
@@ -34,11 +34,14 @@ const (
ReadBuilds = "mesh-delivery-builder.builds" 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 is the vault's bcrypt hashes of the bus accounts it holds, by bus user (ADR 0547 clause 9).
ReadAccounts = "mesh-vault.bus-accounts" 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. // Reads is every state the part reads, in the manifest's order.
var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds, 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. // The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
const ( const (
@@ -103,8 +106,12 @@ type resolved struct {
type Inputs struct { type Inputs struct {
// Nodes is every machine seen: in a node agent's have or a standing assignment. // Nodes is every machine seen: in a node agent's have or a standing assignment.
Nodes []string 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 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 is each module's manifest at the indexer's trunk, by module.
Manifests map[string]catalogue.Manifest Manifests map[string]catalogue.Manifest
// Digests is each module's manifest digest, by module. // Digests is each module's manifest digest, by module.
@@ -230,6 +237,38 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
in.BusNode = a.Node 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) haveKeys, err := r.Keys(ReadHave)
if err != nil { if err != nil {
in.Gaps = append(in.Gaps, Gap{Input: ReadHave, Why: "refused or absent: " + err.Error(), Failed: true}) in.Gaps = append(in.Gaps, Gap{Input: ReadHave, Why: "refused or absent: " + err.Error(), Failed: true})
@@ -7,6 +7,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/buscompose" "git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue" "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, // 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 continue
} }
d := declaredFor(m, seats, declarers) 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) { for _, sr := range catalogue.ReadsGranted(m, here, node) {
d.KeyedReads = append(d.KeyedReads, buscompose.KeyedRead{Read: sr.Read, Key: sr.Key}) 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 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
}
+1
View File
@@ -10,6 +10,7 @@
"mesh-delivery-builder.builds", "mesh-delivery-builder.builds",
"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules", "mesh-delivery-indexer.modules",
"mesh-delivery-indexer.placement",
"node-engine.have", "node-engine.have",
"mesh-vault.bus-accounts" "mesh-vault.bus-accounts"
], ],