novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/main' into fix/delivery-records-true

Browse Source 
# Conflicts:
#	modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
#	modules/mesh-delivery-indexer/go.mod
#	modules/mesh-delivery-indexer/go.sum
#	modules/mesh-delivery-indexer/manifest_test.go
#	modules/mesh-delivery-indexer/module.json
#	modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go
#	modules/mesh-delivery-view/go.mod
#	modules/mesh-delivery-view/go.sum
#	modules/mesh-delivery-view/internal/view/read.go
#	modules/mesh-delivery-view/manifest_test.go
#	modules/mesh-delivery-view/module.json
This commit is contained in:
jschoubben
2026-10-26 22:20:45 +01:00
parent 68173de8bd 4b36cc29a6
commit 971cdb44e2
50 changed files with 1514 additions and 79 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+3
View File
@@ -1 +1,4 @@
vendor/
# A compiled program never enters a commit (issue 1577): `go build .` in a command directory leaves one there.
modules/*/cmd/*/mesh-*
!modules/*/cmd/*/mesh-*.go
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go
+57
View File
@@ -172,6 +172,9 @@ func TestTheAssignerStartsAndSaysWhatItKeeps(t *testing.T) {
// 1297/B2b: a bucket it reads that the bus does not hold stops the start, naming the bucket.
func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
for _, missing := range readBuckets {
if missing == delivery.PlacementBucket {
continue // it starts without it, below
}
b := startBed(t, missing)
select {
case err := <-b.start(t):
@@ -218,6 +221,34 @@ func TestTheAssignerDoesNotStartWithoutABucketItReads(t *testing.T) {
t.Fatal("the put was not taken further once its bucket was made")
}
}
// Review of L2 (D1): without the indexer's placement state it starts all the same, and once the bucket is made (the
// indexer and the assigner come in one delivery) it is read with no restart: a module only the placement names is
// said as placed by one source only.
pb := startBed(t, delivery.PlacementBucket)
pdone := pb.start(t)
select {
case err := <-pdone:
t.Fatalf("without %s the assigner stopped: %v", delivery.PlacementBucket, err)
case <-time.After(2 * time.Second):
}
placement, err := pb.js.CreateKeyValue(pb.ctx, jetstream.KeyValueConfig{Bucket: delivery.PlacementBucket, History: 5})
if err != nil {
t.Fatal(err)
}
pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "n1", Modules: []delivery.PlacedModule{{Module: "web"}}})
plkey, _ := delivery.PlacementKey("n1")
if _, err := placement.Put(pb.ctx, plkey, pl); err != nil {
t.Fatal(err)
}
for deadline := time.Now().Add(15 * time.Second); ; time.Sleep(200 * time.Millisecond) {
if c, ok := pb.conditionsSaid(t)[delivery.AssignerModule+".placement.n1.web"]; ok && c.State == asks.StateRaised {
break
}
if time.Now().After(deadline) {
t.Fatal("the placement state made after the start was not read")
}
}
}
// putJSON writes a value into one of the bed's buckets.
@@ -406,6 +437,32 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
if !cleared {
t.Errorf("the condition of web on neither list on g14 was not cleared once g14 got its want")
}
// Issue 1126 (ADR 0547 §2): the catalogue's placement reaches the indexer's state naming web on every node the
// assignments name it on. Each node then composes its own: its want names the build alone (Self, no composition), and
// its own want says it composes its own, which the node agent reads (mesh-host's TestNats...ComposesAndAppliesItsOwn).
for _, node := range []string{"ace", "g14", "shanks"} {
pl, err := delivery.EncodePlacement(delivery.Placement{Node: node, Modules: []delivery.PlacedModule{{Module: "web"}}})
if err != nil {
t.Fatal(err)
}
pkey, _ := delivery.PlacementKey(node)
b.putJSON(t, delivery.PlacementBucket, pkey, pl)
}
if w := b.wantNow(t, "ace", "web", "ace's want once placed: the build alone, composed by the node", func(w delivery.Want) bool { return w.Self }); w.Composed != "" || w.FP != fNew {
t.Errorf("ace's want once placed = %+v", w)
}
nkey, _ := delivery.WantNodeKey("ace")
var self bool
for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && !self; time.Sleep(100 * time.Millisecond) {
if e, err := b.buckets[delivery.AssignerWantsBucket].Get(b.ctx, nkey); err == nil {
nw, err := delivery.DecodeNodeWant(e.Value())
self = err == nil && nw.Self
}
}
if !self {
t.Error("ace's own want does not say it composes its own once placed")
}
}
// 1297/B9 (the assigner's half, ADR 0528 clause 6): started whole, the assigner holds a fingerprint whose health check
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go
+33 -3
View File
@@ -59,6 +59,7 @@ var readBuckets = []string{
delivery.CandidatesBucket,
delivery.ManifestsBucket,
delivery.ModulesBucket,
delivery.PlacementBucket,
delivery.PresenceBucket,
delivery.HaveBucket,
delivery.StepsBucket(goBuildModule),
@@ -98,6 +99,12 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
logf("the bucket %s cannot be opened (%v): no put is taken further until the assigner starts again", name, err)
continue
}
if name == delivery.PlacementBucket && err != nil {
// The indexer's placement state comes with the second milestone (ADR 0549 §2): without it the controller's
// assignments alone name where modules run, as before any placement was read (ADR 0549 §5).
logf("the bucket %s cannot be opened (%v): the controller's assignments alone are read until the assigner starts again", name, err)
continue
}
if err != nil {
return nil, process.EssentialErr("assigner", name, "write a want from the whole mesh", err)
}
@@ -151,7 +158,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
logger.Printf("what the assigner keeps could not be said on the bus, so its readers are not granted until it is: %v", err)
}
logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
return follow(ctx, js, kvs, sayVia(ctx, js), emitting{said}, logger)
return follow(ctx, js, conn, kvs, sayVia(ctx, js), emitting{said}, logger)
}
// emitting is the process connection as the puts' caller: its calls, and its events with no options.
@@ -169,6 +176,7 @@ var watched = []string{
delivery.BuilderBuildsBucket,
delivery.CandidatesBucket,
delivery.ModulesBucket,
delivery.PlacementBucket,
delivery.PresenceBucket,
delivery.HaveBucket,
delivery.AssignerAsksBucket,
@@ -179,7 +187,7 @@ var watched = []string{
// follow watches every bucket a pass reads and makes a pass at start, once every watch handed what stands, and after
// each change; changes that come during a pass make one more. It returns when ctx ends or a watch ends.
func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(asks.Condition) error, bus caller, logger *log.Logger) error {
func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, bus caller, logger *log.Logger) error {
w := newWorld()
dirty := make(chan struct{}, 1)
wake := func() {
@@ -263,7 +271,12 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
}
logger.Printf("the bucket %s is open: puts are taken further", delivery.AssignerPutsBucket)
}
started := 0
for _, name := range watched {
if kvs[name] == nil && name == delivery.PlacementBucket {
continue // not on the bus at this start (open said so)
}
started++
var opts []jetstream.WatchOpt
if name == delivery.HaveBucket {
opts = append(opts, jetstream.IncludeHistory()) // what each node ran healthy before, for a set-back
@@ -275,7 +288,20 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil {
return err
}
for range len(watched) + 1 {
// The indexer's placement state arrives with the same delivery as this assigner (ADR 0549 §2): when it is not on the
// bus at this start it is bound once the bus says it was created, and watched from then on, with no restart (issue
// 1790's kind). Until then the controller's assignments alone name where modules run (ADR 0549 §5).
placementAppeared := make(chan jetstream.KeyValue, 1)
if kvs[delivery.PlacementBucket] == nil && js != nil {
process.AwaitBucket(ctx, js, nc, delivery.PlacementBucket, "the controller's assignments alone are read",
logger.Printf).Ready(func(kv jetstream.KeyValue) { placementAppeared <- kv })
}
takePlacement := func(kv jetstream.KeyValue) error {
kvs[delivery.PlacementBucket] = kv
logger.Printf("the bucket %s is open: placement is read beside the controller's assignments", delivery.PlacementBucket)
return watch(delivery.PlacementBucket)
}
for range started + 1 {
select {
case <-ctx.Done():
return nil
@@ -297,6 +323,10 @@ func follow(ctx context.Context, js jetstream.JetStream, kvs buckets, say func(a
case err := <-errs:
return err
case <-settled:
case kv := <-placementAppeared:
if err := takePlacement(kv); err != nil {
return err
}
case <-dirty:
reopenPuts(p)
p.pass(ctx)
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go
+1 -1
View File
@@ -94,7 +94,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
kvs[delivery.ControllerComposedBucket].(*watchKV).entries = []kvEntry{{key: ckey, value: cd}}
kvs[delivery.AssignerPutsBucket].(*watchKV).closed = true
kvs[delivery.AssignerWantsBucket].(*watchKV).wrote = wrote
go follow(ctx, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0))
go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, nil, log.New(io.Discard, "", 0))
select {
case key := <-wrote:
if key != "want.ace.node" {
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass.go
+110 -1
View File
@@ -60,7 +60,23 @@ func (p *passer) pass(ctx context.Context) {
// controller has composed it and again whenever that digest changes (ADR 0539 §6).
func (p *passer) nodeWants(ctx context.Context) {
written := p.w.read(delivery.AssignerWantsBucket)
placed := p.w.placedNodes()
// A node the catalogue places composes its own composition (ADR 0547 §2): its own want names none.
for node := range placed {
key, err := delivery.WantNodeKey(node)
if err != nil {
continue
}
if before, err := delivery.DecodeNodeWant(written[key]); err == nil && before.Self {
continue
}
want := delivery.NodeWant{Self: true, Words: "the node composes its own composition (ADR 0547 §2)"}
p.put(ctx, delivery.AssignerWantsBucket, key, func() ([]byte, error) { return delivery.EncodeNodeWant(want) })
}
for node, d := range p.w.nodeComposed() {
if placed[node] {
continue
}
key, err := delivery.WantNodeKey(node)
if err != nil {
continue
@@ -97,6 +113,8 @@ func (p *passer) wants(ctx context.Context) {
}
}
wanted := p.w.wantsOf()
frozen := p.sayPlacement(p.w.placementDiffers())
placed := p.w.placedNodes()
puts := p.putsByModule()
p.staleAsks(ctx, trunk)
names := map[string]bool{}
@@ -119,7 +137,14 @@ func (p *passer) wants(ctx context.Context) {
fp := trunk[name].FP2
name := name
m := assign.Module{Name: name, FP: fp, Manifest: manifests[name], Nodes: nodesOf[name], Now: time.Now(), Puts: puts[name],
Composed: func(node, fp string) (delivery.ComposedDigest, bool) { return p.w.composed(node, name, fp) },
Composed: func(node, fp string) (delivery.ComposedDigest, bool) {
if placed[node] {
// The node composes its own (ADR 0547 §2, §4): no composition of the controller's is waited for, the want
// names the build alone, and the node judges the move itself.
return delivery.ComposedDigest{Judgement: delivery.JudgedNeither, Words: "the node composes its own"}, true
}
return p.w.composed(node, name, fp)
},
GrantedAt: func(fp string) bool {
key, err := delivery.ControllerGrantKey(name, fp)
return err == nil && p.w.has(delivery.ControllerGrantsBucket, key)
@@ -153,6 +178,7 @@ func (p *passer) wants(ctx context.Context) {
last[node] = p.w.lastHealthy(node, name, fp)
}
out := assign.Wants(m, nodes, wanted[name], last)
p.yeses(ctx, name, m, placed, wanted[name], &out)
gatherNeither(neither, name, out)
leftNeither(left, m, wanted[name], out)
p.sayNoWant(name, m.Nodes, wanted[name], out)
@@ -183,6 +209,9 @@ func (p *passer) wants(ctx context.Context) {
p.recordAsk(ctx, name, fp, need)
}
for node, want := range out.Wants {
if frozen[node+"."+name] {
continue // one source alone names it (ADR 0549 §5): no want, and the one written before stays
}
key, err := delivery.WantKey(node, name)
if err != nil {
p.logf("no want of %s on %s is written: %v", name, node, err)
@@ -728,3 +757,83 @@ func (p *passer) backupAsk(ctx context.Context, key, node, module string, want d
}
p.said[c.Key] = true
}
// sayPlacement raises, once, the condition of every pair the controller's assignments and the catalogue's placement do not
// both name, clears it once they agree, and answers those pairs by node and module (ADR 0549 §5): the pass writes no want
// for them and keeps the one written before. A say that fails is made again at the next pass.
func (p *passer) sayPlacement(differ []assign.Pair) map[string]bool {
frozen := map[string]bool{}
now := map[string]bool{}
for _, d := range differ {
frozen[d.Node+"."+d.Module] = true
c := assign.PlacementCondition(d, true)
now[c.Key] = true
if p.said[c.Key+".raised"] {
continue
}
p.logf("%s on %s is named by %s alone: no want is written for it there until both sources name it (ADR 0549 §5)", d.Module, d.Node, d.Only)
if err := p.say(c); err != nil {
p.logf("the placement condition of %s on %s could not be said, and is said at the next pass: %v", d.Module, d.Node, err)
continue
}
p.said[c.Key+".raised"] = true
}
prefix := delivery.AssignerModule + ".placement."
for key := range p.said {
c, raised := strings.CutSuffix(key, ".raised")
if !raised || !strings.HasPrefix(c, prefix) || now[c] {
continue
}
node, module, _ := strings.Cut(strings.TrimPrefix(c, prefix), ".")
if err := p.say(assign.PlacementCondition(assign.Pair{Node: node, Module: module}, false)); err != nil {
p.logf("the placement condition of %s on %s could not be cleared, and is cleared at the next pass: %v", module, node, err)
continue
}
delete(p.said, key)
}
return frozen
}
// yeses writes a placed node's wants naming the build alone, and sees to the moves a node that composes its own judged a
// person must allow (ADR 0547 §3): for a have of phase
// waiting-yes it raises the record ask once, naming the composition waiting and the node's judgement, and on a yes writes
// the want again naming it; a yes once written is kept while the want stays at its build, so the node is not asked again.
func (p *passer) yeses(ctx context.Context, module string, m assign.Module, placed map[string]bool, before map[string]delivery.Want, out *assign.Outcome) {
for _, node := range m.Nodes {
if !placed[node] {
continue
}
was, had := before[node]
w, written := out.Wants[node]
if !written {
if !had || was.Removed {
continue
}
w = was
}
if w.Removed {
continue
}
yes := ""
if had && was.FP == w.FP {
yes = was.Yes
}
key, err := delivery.HaveKey(node, module)
if err != nil {
continue
}
if h, err := delivery.DecodeHave(p.w.value(delivery.HaveBucket, key)); err == nil && h.Phase == delivery.PhaseWaitingYes && h.Waits != yes {
if m.Allowed != nil && m.Allowed(node, h.Waits) {
yes = h.Waits
} else {
p.recordAsk(ctx, module, w.FP, assign.RecordNeed{Node: node, Digest: h.Waits, Judgement: h.Judgement, Words: h.Detail})
}
}
// A want carried over from before the node composed its own may still name the controller's composition; a placed
// node's want names the build alone (ADR 0547 §2).
if yes != w.Yes || w.Composed != "" || !w.Self {
w.Yes, w.Composed, w.Self = yes, "", true
out.Wants[node] = w
}
}
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/pass_test.go
+138 -2
View File
@@ -342,6 +342,9 @@ func putOfTheBus(t *testing.T) {
// again after a say that failed, and clears it once when the module has a want again; a module on neither list is also written to
// the journal, so it is loud with no operator channel.
func TestAModuleOnNeitherListIsRaisedOnceAndClearedOnce(t *testing.T) {
// A pair only one source names is a condition of the same kind (issue 1126): folded here to hold the count.
aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t)
t.Run("gathers", testThePassGathersTheModulesOnNeitherList)
p := newPasser(buckets{}, time.Date(2026, 10, 25, 15, 0, 0, 0, time.UTC))
var said []asks.Condition
failing := true
@@ -391,7 +394,7 @@ func TestAModuleOnNeitherListIsRaisedOnceAndClearedOnce(t *testing.T) {
}
// 1945/AC21: the pass gathers the modules on neither list of every module's outcome, keyed by node and module.
func TestThePassGathersTheModulesOnNeitherList(t *testing.T) {
func testThePassGathersTheModulesOnNeitherList(t *testing.T) {
got := map[string]neitherOne{}
gatherNeither(got, "screen-lock", assign.Outcome{Neither: map[string]assign.Holding{"g14": {Rule: delivery.RuleGrant, Words: "w"}}})
gatherNeither(got, "rofi", assign.Outcome{})
@@ -449,6 +452,7 @@ func TestThePassGathersTheModulesOnNeitherList(t *testing.T) {
// nothing it said, so it clears, once, the neither condition of every assigned module not on neither list and the held condition of
// every module that is not held. Clearing a condition that is not open is quiet at the router.
func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) {
t.Run("a held module", testAHeldModuleRaisesNoNeitherAndClearsAStandingOne)
wantsKV := &memKV{values: map[string][]byte{}}
p := newPasser(buckets{delivery.AssignerWantsBucket: wantsKV, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now())
var said []asks.Condition
@@ -486,7 +490,7 @@ func TestTheFirstPassAfterAStartClearsWhatAPreviousProcessRaised(t *testing.T) {
// 1945/AC24: given a module held at F under rule 2 on a node that runs it with no want, when the assigner passes, then it raises no
// neither condition for it, a neither condition this process raised for it before is cleared, and so is one a previous process
// raised (the first pass after a start), each once.
func TestAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) {
func testAHeldModuleRaisesNoNeitherAndClearsAStandingOne(t *testing.T) {
setup := func() (*passer, *[]asks.Condition) {
p := newPasser(buckets{delivery.AssignerWantsBucket: &memKV{values: map[string][]byte{}}, delivery.AssignerAsksBucket: &memKV{values: map[string][]byte{}}}, time.Now())
said := &[]asks.Condition{}
@@ -702,6 +706,8 @@ func mustKey(key string, err error) string {
// (ADR 0545). That record ask is not about the trunk's fingerprint: it survives every pass, so the operator is asked, and on a
// yes the want names the revised digest.
func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing.T) {
// A placed node's yes comes from its have (issue 1126): folded here to hold the count.
aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t)
fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64)
revised, was := "sha256:"+strings.Repeat("e", 64), "sha256:"+strings.Repeat("d", 64)
asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
@@ -750,3 +756,133 @@ func TestARecordAskOfAHeldBackNodeSurvivesPassesAndItsYesMovesTheWant(t *testing
t.Error("the answered record ask was taken away")
}
}
// 1126 (ADR 0549 §5): until the third milestone a pair the controller's assignments and the catalogue's placement do not
// both name gets no want and one condition, raised once and cleared once both name it; before any placement was read the
// assignments are the one source and nothing is raised.
func aPairOneSourceNamesAloneGetsNoWantAndOneCondition(t *testing.T) {
p := newPasser(buckets{}, time.Date(2026, 10, 26, 18, 0, 0, 0, time.UTC))
var said []asks.Condition
p.say = func(c asks.Condition) error { said = append(said, c); return nil }
assigned := func(node, module string) {
raw, _ := json.Marshal(map[string]string{"node": node, "module": module})
p.w.put(provider.AssignmentsBucket, node+"."+module, raw)
}
place := func(node string, modules ...string) {
pl := delivery.Placement{Node: node}
for _, m := range modules {
pl.Modules = append(pl.Modules, delivery.PlacedModule{Module: m})
}
raw, err := delivery.EncodePlacement(pl)
if err != nil {
t.Fatal(err)
}
key, _ := delivery.PlacementKey(node)
p.w.put(delivery.PlacementBucket, key, raw)
}
assigned("g14", "a")
assigned("g14", "b")
if got, _ := p.w.assigned(); !slices.Equal(got["g14"], []string{"a", "b"}) || len(p.sayPlacement(p.w.placementDiffers())) != 0 || len(said) != 0 {
t.Fatalf("before any placement: assigned %v, said %v", got, said)
}
none := func(context.Context, string) ([]byte, error) { return nil, errNotThere }
if m := p.w.mesh(context.Background(), none); m.Input != nil || m.InputUnread != "" {
t.Errorf("before any placement the check composes over %+v (%q)", m.Input, m.InputUnread)
}
place("g14", "a", "c")
got, _ := p.w.assigned()
frozen := p.sayPlacement(p.w.placementDiffers())
p.sayPlacement(p.w.placementDiffers())
if !slices.Equal(got["g14"], []string{"a"}) || !frozen["g14.b"] || !frozen["g14.c"] || frozen["g14.a"] || len(said) != 2 ||
said[0].State != asks.StateRaised || !strings.Contains(said[0].Explanation, "the controller's assignments names b on g14") ||
!strings.Contains(said[1].Explanation, "the catalogue's placement names c on g14") {
t.Fatalf("one source alone: assigned %v, frozen %v, said %+v", got, frozen, said)
}
place("g14", "a", "b", "c")
assigned("g14", "c")
if frozen := p.sayPlacement(p.w.placementDiffers()); len(frozen) != 0 || len(said) != 4 || said[2].State != asks.StateCleared || said[3].State != asks.StateCleared {
t.Errorf("both agree again: frozen %v, said %+v", frozen, said)
}
// 1126 (ADR 0547 §2): placed, the check composes over the input the nodes compose from (frombus), once each placed node
// wrote its record; until then it says why and composes the assignments alone.
if m := p.w.mesh(context.Background(), none); m.Input != nil || !strings.Contains(m.InputUnread, "g14") {
t.Errorf("a placed node with no record: input %+v, unread %q", m.Input, m.InputUnread)
}
rec, _ := delivery.EncodeHaveNode(delivery.HaveNode{Record: &delivery.NodeRecord{Node: "g14", TunnelKey: "k"}})
p.w.put(delivery.HaveBucket, "g14.node", rec)
m := p.w.mesh(context.Background(), none)
if m.Input == nil || len(m.Input.Nodes) != 1 || !slices.Equal(m.Input.Nodes[0].Assigned, []string{"a", "b", "c"}) || m.Input.Nodes[0].Place.Key != "k" {
t.Errorf("the check's input from the bus: %+v (%q)", m.Input, m.InputUnread)
}
}
// 1126 (ADR 0547 §2 to §4): a node the catalogue places composes its own: its wants name the build alone and its own want
// names no composition; a have waiting for a yes raises the record ask once, and the yes is written into the want and
// kept there while the want stays at its build.
func aPlacedNodesWantsNameTheBuildAloneAndItsYesComesFromItsHave(t *testing.T) {
fpOld := delivery.SourcePrefix2 + strings.Repeat("a", 64)
waiting, was := "sha256:"+strings.Repeat("e", 64), "sha256:"+strings.Repeat("d", 64)
asksKV, wantsKV := &memKV{values: map[string][]byte{}}, &memKV{values: map[string][]byte{}}
p := newPasser(buckets{delivery.AssignerAsksBucket: asksKV, delivery.AssignerWantsBucket: wantsKV}, time.Now())
p.logf = t.Logf
put := func(bucket, key string, v []byte) { p.w.put(bucket, key, v) }
mod, _ := delivery.EncodeModule(delivery.Module{FP2: fpTest, Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
put(delivery.ModulesBucket, "web", mod) // the trunk's build is not done: rule 1 holds g14 at fpOld
put(provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"b-1"}`))
pl, _ := delivery.EncodePlacement(delivery.Placement{Node: "g14", Modules: []delivery.PlacedModule{{Module: "web"}}})
put(delivery.PlacementBucket, "placement.g14", pl)
on, _ := delivery.EncodePresence(delivery.Presence{Online: true})
put(delivery.PresenceBucket, "g14", on)
have, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Composed: was, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
put(delivery.HaveBucket, "g14.web", have)
wkey, _ := delivery.WantKey("g14", "web")
want, _ := delivery.EncodeWant(delivery.Want{FP: fpOld, Composed: was})
wantsKV.values[wkey] = want
put(delivery.AssignerWantsBucket, wkey, want)
sync := func() {
for k, v := range wantsKV.values {
put(delivery.AssignerWantsBucket, k, v)
}
}
p.wants(context.Background())
p.nodeWants(context.Background())
sync()
got, err := delivery.DecodeWant(wantsKV.values[wkey])
if err != nil || got.FP != fpOld || got.Composed != "" || !got.Self {
t.Fatalf("the placed node's want is %+v (%v), want it at %s, the node composing it", got, err, fpOld)
}
nkey, _ := delivery.WantNodeKey("g14")
if nw, err := delivery.DecodeNodeWant(wantsKV.values[nkey]); err != nil || nw.Composed != "" || !nw.Self {
t.Fatalf("the placed node's own want is %+v (%v), want one naming no composition", nw, err)
}
waits, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Composed: was, Phase: delivery.PhaseWaitingYes, Health: delivery.HealthOK,
Judgement: delivery.JudgedRecreates, Waits: waiting, Detail: "it recreates web.server"})
put(delivery.HaveBucket, "g14.web", waits)
akey, _ := delivery.RecordAskKey("g14", "web", waiting)
p.wants(context.Background())
if _, there := asksKV.values[akey]; !there {
t.Fatal("a have waiting for a yes raised no record ask")
}
a, err := delivery.DecodeAssignerAsk(akey, asksKV.values[akey])
if err != nil {
t.Fatal(err)
}
put(delivery.AssignerAsksBucket, akey, asksKV.values[akey])
yes, _ := delivery.EncodeAnswer(delivery.Answer{Option: assign.OptionYes, Warrant: a.Ask.ID + "@2026-10-26T19:00:00Z", Words: "the operator"})
put(delivery.AnswersBucket, akey, yes)
p.wants(context.Background())
sync()
if got, err = delivery.DecodeWant(wantsKV.values[wkey]); err != nil || got.Yes != waiting || got.FP != fpOld {
t.Fatalf("after the yes the want is %+v (%v), want it to name the yes %s", got, err, waiting)
}
// The node applied it: its have runs the composition; the yes stays while the want stays at its build.
ran, _ := delivery.EncodeHave(delivery.Have{FP: fpOld, Composed: waiting, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
put(delivery.HaveBucket, "g14.web", ran)
p.wants(context.Background())
if got, _ = delivery.DecodeWant(wantsKV.values[wkey]); got.Yes != waiting {
t.Errorf("a pass after the move dropped the yes: %+v", got)
}
}
modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go
+128 -1
View File
@@ -14,6 +14,8 @@ import (
"github.com/nats-io/nats.go/jetstream"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
"git.novox.be/novox/mesh-sdk/go/compose/frombus"
"git.novox.be/novox/mesh-sdk/go/delivery"
"git.novox.be/novox/mesh-sdk/go/provider"
@@ -162,8 +164,56 @@ func (w *world) manifest(ctx context.Context, get getter, digest string) parsed
return p
}
// assigned is each node's modules with an open assignment, read from the controller's assignments, in name order.
// assigned is each node's modules both the controller's assignments and the indexer's placement name, in name order
// (ADR 0549 §5; the assignments alone before any placement was read), with every open assignment by node and module.
func (w *world) assigned() (map[string][]string, map[string]assignment) {
byAssignment, all := w.assignedByController()
placed, seeded := w.placed()
both, _ := assign.Placed(byAssignment, placed, seeded)
return both, all
}
// placementDiffers is every pair the controller's assignments and the indexer's placement do not both name.
func (w *world) placementDiffers() []assign.Pair {
byAssignment, _ := w.assignedByController()
placed, seeded := w.placed()
_, differ := assign.Placed(byAssignment, placed, seeded)
return differ
}
// placed is each node's modules in the indexer's placement (ADR 0549 §2), and whether any placement key was read: a key
// that does not read places nothing on its node and is said by the indexer's own read tool.
func (w *world) placed() (map[string][]string, bool) {
out := map[string][]string{}
values := w.read(delivery.PlacementBucket)
for key, value := range values {
node, err := delivery.ParsePlacementKey(key)
if err != nil {
continue
}
p, err := delivery.DecodePlacement(value)
if err != nil || p.Node != node {
continue
}
out[node] = p.Assigned()
}
return out, len(values) > 0
}
// placedNodes is every node the indexer's placement names (ADR 0549 §2): those compose their own modules (ADR 0547 §2),
// so their wants name the build alone and the seventh rule does not hold them.
func (w *world) placedNodes() map[string]bool {
placed, _ := w.placed()
out := make(map[string]bool, len(placed))
for node := range placed {
out[node] = true
}
return out
}
// assignedByController is each node's modules with an open assignment, read from the controller's assignments, in
// name order.
func (w *world) assignedByController() (map[string][]string, map[string]assignment) {
out, all := map[string][]string{}, map[string]assignment{}
for _, value := range w.read(provider.AssignmentsBucket) {
var a assignment
@@ -210,9 +260,86 @@ func (w *world) mesh(ctx context.Context, manifests getter) assign.Mesh {
}
m.Manifests[name] = p.manifest
}
// Once the catalogue places the nodes, the check composes over the input they compose from (ADR 0547 §2, ADR 0549
// §3): one composer's input for the check and the nodes. An input that cannot be read is said, and the nodes and their
// assignments alone are composed, as before any placement.
if _, seeded := w.placed(); seeded {
in, err := frombus.Input(ctx, worldReader{w: w, manifests: manifests}, frombus.Mesh{}, frombus.Scope{Every: true})
if err != nil {
m.InputUnread = err.Error()
} else {
m.Input = &in
}
}
return m
}
// worldReader is what the assigner watched as frombus reads the bus: the indexer's placement and modules, each node's
// record in its own have key, the manifests by digest; the check reads no config value, own secret or credential (the
// composition it asks resolves, and stands in for what it does not carry).
type worldReader struct {
w *world
manifests getter
}
func (r worldReader) Placements(context.Context) (map[string]delivery.Placement, error) {
out := map[string]delivery.Placement{}
for key, value := range r.w.read(delivery.PlacementBucket) {
node, err := delivery.ParsePlacementKey(key)
if err != nil {
continue
}
if p, err := delivery.DecodePlacement(value); err == nil && p.Node == node {
out[node] = p
}
}
return out, nil
}
func (r worldReader) Records(context.Context) (map[string]delivery.NodeRecord, error) {
out := map[string]delivery.NodeRecord{}
for key, value := range r.w.read(delivery.HaveBucket) {
node, _, isNode, err := delivery.ParseHaveKey(key)
if err != nil || !isNode {
continue
}
if h, err := delivery.DecodeHaveNode(value); err == nil && h.Record != nil {
out[node] = *h.Record
}
}
return out, nil
}
func (r worldReader) Modules(context.Context) (map[string]delivery.Module, error) {
return r.w.trunk(), nil
}
func (r worldReader) Manifest(ctx context.Context, digest string) ([]byte, error) {
key, err := delivery.ManifestsKey(digest)
if err != nil {
return nil, err
}
raw, err := r.manifests(ctx, key)
if err != nil {
return nil, err
}
m, err := delivery.DecodeManifest(raw)
if err != nil {
return nil, err
}
return []byte(m.JSON), nil
}
func (worldReader) Values(context.Context, string, string) (map[string]any, error) { return nil, nil }
func (worldReader) OwnSecrets(context.Context, string) ([]compose.OwnSecretRow, error) {
return nil, nil
}
func (worldReader) Credentials(context.Context, string, []compose.CredentialKey, []compose.CredentialKey) ([]compose.Credential, []compose.Issued, error) {
return nil, nil, nil
}
// change is a candidate's change for the composition check, or why the composition waits for it.
func (w *world) change(ctx context.Context, manifests getter, c delivery.Candidate, trunk map[string]delivery.Module) (assign.Change, string) {
ch := assign.Change{Group: c.Group, Heads: c.HeadsHash, Manifests: map[string]catalogue.Manifest{}, Unreadable: map[string]string{},
modules/mesh-delivery-assigner/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-assigner
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-assigner/go.sum
+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.20261024210118-748c564cdf7c/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025195051-20a6dc4920f8 h1:jA7V7hlfJAigVUvJDpKwfgd9ylQ9bqppg3rK9XgLWoA=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025195051-20a6dc4920f8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8 h1:vhYk/iv43u2WGSTrFyQGDvGBxLaRITgMAiiI9DjuA10=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025200810-d4ae4b6790b8/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5 h1:gE80bR0tPME7MTMBLcdP62v+F/9soVXmdcBYNMOye44=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025203528-db26f7e8b6f5/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-delivery-assigner/internal/assign/compose.go
+7
View File
@@ -35,6 +35,13 @@ type Mesh struct {
// Digests is every trunk module's manifest digest in the indexer's manifests bucket, readable or not: what a
// composer at another mesh-sdk reads (issue 1622).
Digests map[string]string
// Input is the rest of the mesh as the nodes compose from it, filled by mesh-sdk's frombus from the catalogue's
// placement and the nodes' records (novox/hq ADR 0547 §2, ADR 0549 §3, issue 1126), so the check and the nodes share
// one composer's input; nil before any placement is read, when the nodes and their assignments alone are composed.
Input *compose.Input
// InputUnread says why that input could not be read from the bus, when it could not: the check then composes the
// nodes and their assignments alone, and says so in its warnings.
InputUnread string
}
// Change is a group's change as the indexer's candidate holds it: the manifest of every module it moves or adds, those
modules/mesh-delivery-assigner/internal/assign/composer.go
+13 -2
View File
@@ -42,8 +42,15 @@ func HeadJob(m Mesh, c Change) (delivery.ComposeJob, error) {
for _, name := range c.Removed {
delete(change, name)
}
req := composer.Request{V: composer.Version, Input: plainInput(m.Assigned), Depth: composer.DepthResolve,
HoldingsFromClaims: true, Machines: Touched(m, c),
in, fromClaims := plainInput(m.Assigned), true
if m.Input != nil {
// The input the nodes compose from, without its catalogue: the cases carry the manifests (issue 1126).
in = *m.Input
in.Catalogue = nil
fromClaims = len(in.Holdings) == 0
}
req := composer.Request{V: composer.Version, Input: in, Depth: composer.DepthResolve,
HoldingsFromClaims: fromClaims, Machines: Touched(m, c),
Cases: []composer.Case{{Name: CaseTrunk, Manifests: trunk}, {Name: CaseChange, Manifests: change}}}
return job(SDKRepository, c.SDKHead, SDKDir, req, "the mesh-sdk head of "+c.Group)
}
@@ -195,6 +202,10 @@ func FromComposer(m Mesh, c Change, head Ran, readers []Reader, took time.Durati
readerFails, readerWarnings := judgeReaders(c, change, readers)
fails = append(fails, readerFails...)
v.Warnings = append(v.Warnings, readerWarnings...)
if m.InputUnread != "" {
v.Warnings = append(v.Warnings, "composed over the nodes and their assignments alone: the input the nodes compose from "+
"cannot be read from the bus ("+m.InputUnread+")")
}
return finish(v, m, c, machines, fails)
}
modules/mesh-delivery-assigner/internal/assign/composer_test.go
+20
View File
@@ -8,6 +8,7 @@ import (
"time"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/compose"
"git.novox.be/novox/mesh-sdk/go/composer"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
@@ -100,6 +101,25 @@ func TestAGroupWithAnSDKMemberAsksTheComposerAtItsHead(t *testing.T) {
if !slices.Equal(req.Machines, []string{"g14"}) || req.Depth != composer.DepthResolve || !req.HoldingsFromClaims {
t.Errorf("request = machines %v, depth %q, holdings from claims %v", req.Machines, req.Depth, req.HoldingsFromClaims)
}
// 1126 (ADR 0547 §2, ADR 0549 §3): once the catalogue places the nodes, the check composes over the input the nodes
// compose from (frombus), its catalogue left to the cases, and the seats held as placement says.
m := digestedMesh()
m.Input = &compose.Input{Catalogue: map[string]catalogue.Manifest{"web": {Module: "web"}},
Nodes: []compose.Node{{Name: "g14", Assigned: []string{"web"}, Place: compose.Place{Endpoint: "g14.example:51820"}}},
Holdings: []catalogue.Held{{Claim: "web-seat", Scope: "mesh", Node: "g14", Module: "web"}}}
placedJob, err := HeadJob(m, c)
if err != nil {
t.Fatal(err)
}
placed, err := composer.Decode(placedJob.Request)
if err != nil {
t.Fatal(err)
}
if placed.Input.Catalogue != nil || len(placed.Input.Nodes) != 1 || placed.Input.Nodes[0].Place.Endpoint != "g14.example:51820" ||
len(placed.Input.Holdings) != 1 || placed.HoldingsFromClaims {
t.Errorf("the placed check's request = %+v, holdings from claims %v", placed.Input, placed.HoldingsFromClaims)
}
}
// 1622/AC2, AC3: a catalogue member using a field only the SDK head reads passes with the head's composer, and the
modules/mesh-delivery-assigner/internal/assign/placement.go
+99
View File
@@ -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)}
}
modules/mesh-delivery-assigner/manifest_test.go
+1 -1
View File
@@ -34,7 +34,7 @@ func TestManifest(t *testing.T) {
slices.Sort(reads)
want := []string{"mesh-build-go.steps", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants",
"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
"mesh-delivery-indexer.modules", "mesh-delivery-indexer.placement", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "node-engine.have"}
if !reflect.DeepEqual(reads, want) {
t.Errorf("reads = %v, want exactly %v", reads, want)
}
modules/mesh-delivery-assigner/module.json
+1
View File
@@ -12,6 +12,7 @@
"mesh-delivery-indexer.candidates",
"mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.placement",
"mesh-delivery-teller.answers",
"mesh-delivery-watchman.presence",
"node-engine.have"
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/integration_test.go
+2 -1
View File
@@ -427,7 +427,8 @@ func (b *bed) start(t *testing.T, env map[string]string) {
if err := os.WriteFile(broker, cred, 0o600); err != nil {
t.Fatal(err)
}
e := map[string]string{"MESH_BROKER_FILE": broker}
// The settle window of a group's heads (check.DefaultSettle) is cut short: each bed's candidate is written whole.
e := map[string]string{"MESH_BROKER_FILE": broker, "MESH_CHECKER_SETTLE": "10ms"}
for k, v := range env {
e[k] = v
}
modules/mesh-delivery-checker/cmd/mesh-delivery-checker/main.go
+5
View File
@@ -189,6 +189,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
settleWindow, err := durationEnv(getenv, "MESH_CHECKER_SETTLE", check.DefaultSettle)
if err != nil {
return err
}
link := strings.TrimSpace(getenv("MESH_CHECKER_LINK"))
if link == "" {
link = defaultLink
@@ -221,6 +225,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
Bound: checkBound,
Link: link,
Logf: logger.Printf,
Settle: settleWindow,
}
// The builder's builds say how the candidates' fingerprints stand (issue 1684); where its bucket is not there yet the line
// says nothing of the builds, until the checker starts again.
modules/mesh-delivery-checker/internal/check/checker.go
+176 -17
View File
@@ -85,6 +85,10 @@ type Checker struct {
Announce Announcer
Now func() time.Time
Logf func(format string, args ...any)
// Settle is how long a group's heads must stay the same after the candidate last changed them before its check starts
// (issue 1890): the members of a group are announced one after another, and a check of each partial set is thrown
// away at the next. None when 0; DefaultSettle in the module.
Settle time.Duration
once sync.Once
wg sync.WaitGroup
@@ -96,15 +100,77 @@ type Checker struct {
epoch uint64 // changes of the nodes seen since this process started
nodesFail map[string]uint64 // by verdict key: the epoch a failure of the nodes was written at
held map[string]string // by verdict key: the turn that holds its check off, as last said
live map[string]liveCheck // by verdict key: the check running in this process, ended when its heads go
settling map[string]*window // by candidates key: the window its heads settle in
}
// liveCheck is a check running in this process: the candidates key of its group and the end of its run.
type liveCheck struct {
group string
end context.CancelCauseFunc
}
// window is a group's settle window: aborted with errUnsettled when its heads change again or the group goes, and with
// errSettled to end it at once.
type window struct{ abort context.CancelCauseFunc }
var (
errUnsettled = errors.New("the group's heads changed again")
errSettled = errors.New("the group's heads settled")
)
// settleNow ends the settle window of every group at once, as if Settle had passed.
func (ch *Checker) settleNow() {
ch.mu.Lock()
defer ch.mu.Unlock()
for _, w := range ch.settling {
w.abort(errSettled)
}
}
// DefaultSettle is the module's Settle (issue 1890). On 2026-10-26 the three members of a group were announced within 2.8
// seconds of each other (20:31:52.3, :53.7, :55.1 UTC), a session opening or pushing them one API call after another;
// fifteen seconds holds five times that, and costs a check of 10 to 18 minutes under 2.5 % of its time. A member that comes
// later still is no waste beyond its asks already taken: the check of the heads it replaced is ended at once.
const DefaultSettle = 15 * time.Second
// EndedWords open the words of a verdict the checker ended without judging it, its heads no candidate's any more (issue
// 1890): never a failure of the change, and judged whole if a candidate holds those heads again.
const EndedWords = "ended: "
// IsEnded says whether a verdict was ended without being judged (EndedWords).
func IsEnded(v delivery.Verdict) bool {
return v.State == delivery.VerdictFail && v.Failed.Machine == "checker" && strings.HasPrefix(v.Failed.Words, EndedWords)
}
// superseded is the cause a running check is ended with: the heads that replaced its own, or none when its group went.
type superseded struct{ by string }
func (s superseded) Error() string {
if s.by == "" {
return EndedWords + "no candidate holds these heads any more: the group was closed or merged"
}
return EndedWords + "superseded by " + s.by
}
func (ch *Checker) init() {
ch.once.Do(func() {
ch.known, ch.inflight, ch.again, ch.tried = map[string]delivery.Candidate{}, map[string]bool{}, map[string]bool{}, map[string]bool{}
ch.nodesFail, ch.held = map[string]uint64{}, map[string]string{}
ch.live, ch.settling = map[string]liveCheck{}, map[string]*window{}
})
}
// endOthers ends every check of the group running in this process whose verdict key is not keep: no candidate holds its
// heads any more (issue 1890). Called with ch.mu held.
func (ch *Checker) endOthers(group, keep, by string) {
for k, l := range ch.live {
if l.group == group && k != keep {
l.end(superseded{by: by})
}
}
}
func (ch *Checker) logf(format string, args ...any) {
if ch.Logf != nil {
ch.Logf(format, args...)
@@ -137,17 +203,62 @@ func (ch *Checker) Candidate(ctx context.Context, key string, value []byte) {
ch.logf("candidates.%s holds what is not a candidate, left: %v", key, err)
return
}
vkey, _ := delivery.CheckerVerdictKey(c.Group, c.HeadsHash)
ch.mu.Lock()
prev, had := ch.known[key]
ch.known[key] = c
// A check of heads this candidate no longer holds is ended at once (issue 1890): a member's new head, a member joining
// or leaving.
ch.endOthers(key, vkey, GateHeads(c))
moved := !had || prev.HeadsHash != c.HeadsHash
if ch.Settle > 0 && (moved || ch.settling[key] != nil) {
// The group's heads changed: its check waits until they have been the same for Settle, so a group whose members
// are announced one after another is checked once, whole. A later change starts the window again. The wait is a
// context's bound, no timer (ADR 0437 §8).
if w := ch.settling[key]; w != nil {
w.abort(errUnsettled)
}
base, abort := context.WithCancelCause(ctx)
w := &window{abort: abort}
ch.settling[key] = w
ch.wg.Add(1)
go func() {
defer ch.wg.Done()
wait, done := context.WithTimeout(base, ch.Settle)
defer done()
<-wait.Done()
settled := errors.Is(wait.Err(), context.DeadlineExceeded) || errors.Is(context.Cause(wait), errSettled)
ch.mu.Lock()
if !settled || ch.settling[key] != w {
ch.mu.Unlock()
return // changed again, forgotten, or the checker stops
}
delete(ch.settling, key)
now, ok := ch.known[key]
ch.mu.Unlock()
abort(nil)
if ok {
ch.consider(ctx, key, now)
}
}()
ch.mu.Unlock()
return
}
ch.mu.Unlock()
ch.consider(ctx, key, c)
}
// Forget takes a candidate out (its key was deleted).
// Forget takes a candidate out (its key was deleted): a check of its heads still running is ended, as no candidate holds
// them any more.
func (ch *Checker) Forget(key string) {
ch.init()
ch.mu.Lock()
delete(ch.known, key)
if w := ch.settling[key]; w != nil {
w.abort(errUnsettled)
delete(ch.settling, key)
}
ch.endOthers(key, "", "")
ch.mu.Unlock()
}
@@ -190,32 +301,49 @@ func (ch *Checker) consider(ctx context.Context, key string, c delivery.Candidat
}
cs.Queued = ch.now()
ch.mu.Lock()
if ch.settling[key] != nil {
ch.mu.Unlock()
return // its heads are settling: the window's end considers it (Settle)
}
if ch.inflight[cs.Key] {
ch.again[cs.Key] = true // looked at once more when the check ends, so a change of the trunk is never lost
ch.mu.Unlock()
return
}
ch.inflight[cs.Key] = true
run, end := context.WithCancelCause(ctx)
ch.live[cs.Key] = liveCheck{group: key, end: end}
ch.mu.Unlock()
ch.wg.Add(1)
go func() {
defer ch.wg.Done()
vkey := cs.Key
defer func() {
ch.mu.Lock()
delete(ch.inflight, vkey)
delete(ch.again, vkey)
delete(ch.live, vkey)
ch.mu.Unlock()
end(nil)
}()
for {
// Every check is asked at once: the build seats' streams hold the backlog, a new head's check and a check after a
// failure of the nodes on check, a re-judge after a trunk move on recheck, pulled after them (ADR 0525 §4, §9;
// ADR 0536 §4). The checker holds no place and no capacity of its own.
ch.prepare(ctx, cs)
ch.try(ctx, key, cs)
ch.try(ctx, run, key, cs)
ch.mu.Lock()
more := ch.again[cs.Key]
delete(ch.again, cs.Key)
if !more || ctx.Err() != nil {
delete(ch.inflight, cs.Key)
ch.mu.Unlock()
more := ch.again[vkey]
delete(ch.again, vkey)
ch.mu.Unlock()
if !more || run.Err() != nil {
return
}
ch.mu.Unlock()
// Looked at again for the same heads only: a candidate that moved to other heads has its own check (issue 1890).
if next := CaseOf(ch.candidate(key, c), ch.heads()); next != nil {
if next.Key != vkey {
return
}
cs = next
}
cs.Queued = ch.now()
@@ -252,6 +380,8 @@ func (ch *Checker) prepare(ctx context.Context, cs *Case) {
return
case IsConflictFailure(v):
return // a head git refused: its next check is whole (ADR 0548 §7)
case IsEnded(v):
return // ended unjudged when its heads went (issue 1890): held again, they are checked whole
}
cs.Kind = KindTrunk
if v.State == delivery.VerdictPass && ch.stale(v, cs) {
@@ -402,8 +532,8 @@ func (ch *Checker) claim(ctx context.Context, cs *Case, first bool, epoch uint64
return &claimed{rev}, nil
}
return nil, nil
case ch.stale(cur, cs):
retake = true
case IsEnded(cur), ch.stale(cur, cs):
retake = true // heads held again after their check was ended unjudged (issue 1890), or a trunk that moved
case cur.State == delivery.VerdictFail && IsNodesFailure(cur):
ch.mu.Lock()
at, ok := ch.nodesFail[cs.Key]
@@ -483,9 +613,13 @@ func IsNodesFailure(v delivery.Verdict) bool {
return v.State == delivery.VerdictFail && strings.Contains(v.Failed.Words, UnheardWords)
}
// try claims the check, judges it within its bound and writes the verdict and the owed statuses. A check that ends
// because ctx did is left running: the checker takes it again when it starts.
func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
// try claims the check, judges it within its bound under run and writes the verdict and the owed statuses. A check that ends
// because ctx did is left running: the checker takes it again when it starts. A check whose run was ended because no
// candidate holds its heads any more (issue 1890) is ended: its verdict says so, and no status is written.
func (ch *Checker) try(ctx, run context.Context, key string, cs *Case) {
if run.Err() != nil {
return // ended before it was claimed: nothing to write
}
ch.mu.Lock()
first := !ch.tried[cs.Key]
ch.tried[cs.Key] = true
@@ -501,21 +635,26 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
}
began := ch.now()
ch.writeOwed(ctx, cs, delivery.Verdict{State: delivery.VerdictRunning}, Outcome{})
out := ch.judge(ctx, cs)
out := ch.judge(run, cs)
// An ask the build seat gave up on judged nothing (issue 734): the check is asked again under the same claim, its status
// pending, and after LostAskTries it fails as not heard, judged again when the nodes change.
for try := 1; IsLostAsk(out.Verdict) && ctx.Err() == nil; try++ {
for try := 1; IsLostAsk(out.Verdict) && run.Err() == nil; try++ {
if try >= LostAskTries {
out.Verdict.Failed.Words = strings.TrimSpace(out.Verdict.Failed.Words + "\nasked " + fmt.Sprint(LostAskTries) +
" times and given up each time: " + UnheardWords)
break
}
ch.logf("%s: %s; asked again (%d of %d)", cs.Key, out.Verdict.Failed.Words, try+1, LostAskTries)
out = ch.judge(ctx, cs)
out = ch.judge(run, cs)
}
if ctx.Err() != nil {
return
}
var why superseded
if errors.As(context.Cause(run), &why) {
ch.end(ctx, cs, cl, why)
return
}
v := out.Verdict
v.Group, v.Heads, v.Trunk, v.Beside = cs.Group, cs.Heads, cs.Trunk, nil
for r, c := range out.Beside {
@@ -569,6 +708,26 @@ func (ch *Checker) try(ctx context.Context, key string, cs *Case) {
ch.logf("%s: %s", cs.Key, v.State)
}
// end writes the verdict of a check whose heads no candidate holds any more (issue 1890), by compare-and-set on its claim,
// and says it once in the log with the asks it had put: they stay in the build seats' streams, since an ask once put cannot
// be taken back (buildrun.Asker has no withdraw), and are no longer waited for. No status is written: the heads that
// replaced these have their own check, whose pending status stands.
func (ch *Checker) end(ctx context.Context, cs *Case, cl *claimed, why superseded) {
v := delivery.Verdict{Group: cs.Group, Heads: cs.Heads, Trunk: cs.Trunk, State: delivery.VerdictFail,
Failed: delivery.FailedFirst{Machine: "checker", Words: why.Error()}}
data, err := delivery.EncodeVerdict(v)
if err == nil {
_, err = ch.Verdicts.Update(ctx, cs.Key, data, cl.rev)
}
if err != nil {
ch.logf("%s: %s, but the verdict could not say so: %v", cs.Key, why.Error(), err)
return
}
stage, _ := cs.Progress.Now()
asks := cs.Asks()
ch.logf("%s: %s; it was at %q with %d asks put %v, which are no longer waited for", cs.Key, why.Error(), stage, len(asks), asks)
}
// judge runs the case within the bound. A judge that fails to judge, and a check that ran past its bound, are verdicts of
// failure naming the machine and the step.
func (ch *Checker) judge(ctx context.Context, cs *Case) Outcome {
@@ -667,7 +826,7 @@ func (ch *Checker) Sweep(ctx context.Context) error {
continue
}
cur.State = delivery.VerdictFail
cur.Failed = delivery.FailedFirst{Machine: "checker", Words: "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"}
cur.Failed = delivery.FailedFirst{Machine: "checker", Words: EndedWords + "closed at start: left running, and no candidate holds these heads any more, so nothing will check them"}
data, err := delivery.EncodeVerdict(cur)
if err != nil {
return err
modules/mesh-delivery-checker/internal/check/checker_test.go
+85 -2
View File
@@ -3,6 +3,7 @@ package check
import (
"context"
"encoding/json"
"fmt"
"strings"
"sync"
"testing"
@@ -333,7 +334,10 @@ func TestTheStatusIsPendingWhileTheCheckRunsAndEndsSuccessOrFailure(t *testing.T
}
}
func TestSweepClosesARunningKeyNoCandidateHoldsAndLeavesOneItDoes(t *testing.T) {
// No check of heads no candidate holds is left running (issue 1890): at start the sweep ends a key left running; while the
// checker runs, a check whose group moved to other heads is ended at once, unjudged, and writes no status; heads held again
// are checked whole; and a group's check starts only once its heads settled, so a partial set of members is never checked.
func TestHeadsNoCandidateHoldsAreNeverLeftRunning(t *testing.T) {
ctx, j := context.Background(), passJudge()
r := newRig(t, j)
cand := candidateOf(t, "feat/x", pullOf(repoA, 7, head1))
@@ -351,12 +355,91 @@ func TestSweepClosesARunningKeyNoCandidateHoldsAndLeavesOneItDoes(t *testing.T)
if err := r.ch.Sweep(ctx); err != nil {
t.Fatal(err)
}
if v := r.verdict(t, dead); v.State != delivery.VerdictFail || !strings.Contains(v.Failed.Words, "closed at start") {
if v := r.verdict(t, dead); !IsEnded(v) || !strings.Contains(v.Failed.Words, "closed at start") {
t.Errorf("the dead key = %+v", v)
}
if v := r.verdict(t, live); v.State != delivery.VerdictRunning {
t.Errorf("the live key = %+v, want it left for the check to take over", v)
}
// A member joins while the check of the lone member's heads runs: that check is ended, its judge's run cancelled.
started, ended := make(chan struct{}), make(chan error, 1)
j = &judgeFunc{f: func(run context.Context, _ *Case, call int) (Outcome, error) {
if call == 1 {
close(started)
<-run.Done()
ended <- context.Cause(run)
return Outcome{}, run.Err()
}
return Outcome{Verdict: passing()}, nil
}}
r = newRig(t, j)
var logged []string
var logMu sync.Mutex
r.ch.Logf = func(f string, a ...any) { logMu.Lock(); logged = append(logged, fmt.Sprintf(f, a...)); logMu.Unlock() }
alone, both := cand, candidateOf(t, "feat/x", pullOf(repoA, 7, head1), pullOf(repoSDK, 3, head2))
r.ch.Candidate(ctx, "feat%2Fx", alone)
<-started
r.ch.Candidate(ctx, "feat%2Fx", both)
var bothC delivery.Candidate
_ = json.Unmarshal(both, &bothC)
select {
case err := <-ended:
if err == nil || !strings.Contains(err.Error(), "superseded by "+GateHeads(bothC)) {
t.Errorf("the old check's run ended with %v", err)
}
case <-time.After(5 * time.Second):
t.Fatal("the check of the heads the candidate left still runs")
}
r.ch.Wait()
if v := r.verdict(t, live); !IsEnded(v) || v.Failed.Words != EndedWords+"superseded by "+GateHeads(bothC) {
t.Errorf("the superseded verdict = %+v", v.Failed)
}
if v := r.verdict(t, keyOf(t, "feat/x", both)); v.State != delivery.VerdictPass {
t.Errorf("the group's verdict = %s", v.State)
}
okey, _ := delivery.CheckerOwedKey(repoA, 7, head1)
if raw, _, _, _ := r.owed.Get(ctx, okey); !strings.Contains(string(raw), `"success"`) {
t.Errorf("the head's status = %s, want the group's success, never the ended check's", raw)
}
logMu.Lock()
said := 0
for _, l := range logged {
if strings.Contains(l, "ended: superseded") {
said++
}
}
logMu.Unlock()
if said != 1 {
t.Errorf("the end was said %d times: %v", said, logged)
}
// The member leaves again: the ended heads are held again and checked whole.
r.ch.Candidate(ctx, "feat%2Fx", alone)
r.ch.Wait()
if v := r.verdict(t, live); v.State != delivery.VerdictPass || j.count() != 3 {
t.Errorf("the heads held again = %s after %d judgings, want a pass of a third", v.State, j.count())
}
// Members announced one after another: only the settled, whole set is checked.
j = passJudge()
r = newRig(t, j)
r.ch.Settle = time.Hour
sdk := candidateOf(t, "feat/y", pullOf(repoSDK, 4, head2))
whole := candidateOf(t, "feat/y", pullOf(repoSDK, 4, head2), pullOf(repoA, 8, head1))
r.ch.Candidate(ctx, "feat%2Fy", sdk)
r.ch.Candidate(ctx, "feat%2Fy", whole)
r.ch.Reconsider(ctx) // a trunk move or a turn while the heads settle starts nothing
if j.count() != 0 {
t.Fatalf("judged %d times before the heads settled", j.count())
}
r.ch.settleNow()
r.ch.Wait()
if _, _, found, _ := r.verdicts.Get(ctx, keyOf(t, "feat/y", sdk)); found || j.count() != 1 {
t.Errorf("the partial set has a verdict (%v) or the judge ran %d times, want the whole set alone", found, j.count())
}
if v := r.verdict(t, keyOf(t, "feat/y", whole)); v.State != delivery.VerdictPass {
t.Errorf("the whole set = %s", v.State)
}
}
func TestTheForgeBridgeIsToldOnceForEachChangeOfAStatus(t *testing.T) { // 1683 (the checker's half of AC5)
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/call.go
+22
View File
@@ -91,6 +91,28 @@ func treesVia(ctx context.Context, call caller) bridge.Trees {
}
}
// listedVia asks the forge for the files directly under one directory of a commit through call: gitea_commit_trees with
// listed (issue 1126), answer {listed: {directory: {name: blob id}}}.
func listedVia(ctx context.Context, call caller) bridge.Listed {
return func(owner, repo, sha, dir string) (map[string]string, error) {
raw, err := call(ctx, commitTreesTool, map[string]any{"owner": owner, "repo": repo, "sha": sha, "listed": []string{dir}})
if err != nil {
return nil, err
}
var a struct {
Listed map[string]map[string]string `json:"listed"`
}
if err := json.Unmarshal(raw, &a); err != nil {
return nil, fmt.Errorf("the answer of %s is not {listed}: %w", commitTreesTool, err)
}
if a.Listed == nil {
// A forge module from before listed answers none: placement cannot be read, never read as empty.
return nil, fmt.Errorf("the answer of %s lists no directory; the forge module does not list yet", commitTreesTool)
}
return a.Listed[dir], nil
}
}
// getFileTool is the forge's tool for one file at a ref (mesh-catalog modules/gitea tools/index.ts gitea_get_file):
// arguments owner, repo, path and ref, answer {file: {path, ref, sha, size, content}}, the content decoded to text
// (client.ts getFile). A path the forge does not have fails with `Gitea API <path>: 404 <body>` (client.ts request).
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main.go
+3 -3
View File
@@ -117,8 +117,8 @@ type publisher interface {
// handle turns one event into its message on FORGE and acknowledges it. An event it cannot read is terminated, never
// redelivered, and the log names its subject and why. A message it could not write stays unacknowledged: the server
// delivers it again (the server's ack wait is the only clock).
func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, logger *log.Logger) {
msg, err := bridge.Convert(m.Subject(), m.Data(), trees, files)
func handle(ctx context.Context, m delivered, pub publisher, trees bridge.Trees, files bridge.Files, list bridge.Listed, logger *log.Logger) {
msg, err := bridge.ConvertWith(m.Subject(), m.Data(), trees, files, list)
switch {
case errors.Is(err, bridge.ErrRetry):
logger.Printf("%v; heard again in %s", err, retryDelay)
@@ -237,7 +237,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
// Bounded: the turn is held, and no sync or other event goes on without it.
hctx, cancel := context.WithTimeout(ctx, syncDeadline)
defer cancel()
handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), logger)
handle(hctx, m, rec, treesVia(hctx, call), filesVia(hctx, call), listedVia(hctx, call), logger)
})
},
jetstream.ConsumeErrHandler(func(_ jetstream.ConsumeContext, err error) { logger.Printf("hearing the bus: %v", err) }))
modules/mesh-delivery-forge-bridge/cmd/mesh-delivery-forge-bridge/main_test.go
+5 -5
View File
@@ -81,7 +81,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit
})
var logs bytes.Buffer
m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{}
handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0))
handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
if len(pub.subjects) != 1 || pub.subjects[0] != "mesh.stream.mesh-delivery-forge-bridge.event.forge.novox.mesh-delivery.pull" {
t.Fatalf("published on %v", pub.subjects)
}
@@ -97,7 +97,7 @@ func TestReadableEventIsWrittenOnForgeAndAcked(t *testing.T) { // 1235/AC2 (unit
func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5
var logs bytes.Buffer
m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(map[string]any{"head_sha": "short"})}, &fakePub{}
handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0))
handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
if len(pub.subjects) != 0 {
t.Errorf("a message was written: %v", pub.subjects)
}
@@ -112,7 +112,7 @@ func TestUnreadableEventIsTerminatedAndLogged(t *testing.T) { // 1235/AC5
func TestUnwrittenMessageIsNotAcked(t *testing.T) { // 1235 (a failed write is redelivered by the server, not retried here)
var logs bytes.Buffer
m, pub := &fakeMsg{subject: "mesh.mod.gitea.event.pull.updated", data: body(nil)}, &fakePub{err: errors.New("no stream")}
handle(context.Background(), m, pub, nil, someFiles, log.New(&logs, "", 0))
handle(context.Background(), m, pub, nil, someFiles, nil, log.New(&logs, "", 0))
if m.acked != 0 || m.termd != 0 {
t.Errorf("acked %d terminated %d, want neither", m.acked, m.termd)
}
@@ -134,7 +134,7 @@ func TestTreesRefusedLeavesTheEventForTenSecondsAndLogs(t *testing.T) { // 1236/
call := func(context.Context, string, any) (json.RawMessage, error) {
return nil, errors.New("the tree listing of novox/mesh-delivery was cut")
}
handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), log.New(&logs, "", 0))
handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0))
if len(pub.subjects) != 0 {
t.Errorf("a message was written: %v", pub.subjects)
}
@@ -163,7 +163,7 @@ func TestMergeCallsTheToolOnceAndWritesTheMergedMessage(t *testing.T) { // 1236/
}
return json.RawMessage(`{"root":"r0","trees":{"":"t0","modules/a":"t1"}}`), nil
}
handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), log.New(&logs, "", 0))
handle(context.Background(), m, pub, treesVia(context.Background(), call), filesVia(context.Background(), call), nil, log.New(&logs, "", 0))
if len(keys) == 0 || keys[0] != "gitea.gitea_commit_trees" || slices.Index(keys[1:], "gitea.gitea_commit_trees") >= 0 {
t.Fatalf("calls %v, want the trees once and first", keys)
}
modules/mesh-delivery-forge-bridge/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-forge-bridge
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-forge-bridge/go.sum
+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.20261022190740-104533e8c0c9/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585 h1:7QoFH5qjXRpx5M4uPdYsVJ5qKl2082gq8n0lFxhoNIg=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261025211220-e0d1de047585/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-delivery-forge-bridge/internal/bridge/bridge.go
+50 -4
View File
@@ -7,6 +7,7 @@ import (
"encoding/json"
"errors"
"fmt"
"maps"
"path"
"regexp"
"slices"
@@ -88,6 +89,17 @@ func unreadable(subject, why string, args ...any) error {
// the trunk asks trees once and becomes a merged message; a merge into any other branch is a closed pull message, as the
// pull request left its group. A forge that cannot say is ErrRetry; a merge with no module.json is ErrNoModules.
func Convert(subject string, data []byte, trees Trees, files Files) (Message, error) {
return ConvertWith(subject, data, trees, files, nil)
}
// Listed asks the forge for the files directly under one directory of a commit, by name to blob id (gitea_commit_trees
// with listed); a directory the commit lacks lists none.
type Listed func(owner, repo, sha, dir string) (map[string]string, error)
// ConvertWith is Convert; with list, a merge of the catalogue repository (delivery.CatalogueRepository) also carries every
// placement file of its merge commit (novox/hq ADR 0549 §2, issue 1126), read whole or not at all: a forge that cannot
// say is ErrRetry, as for the manifests, so the indexer never takes a part of placement for all of it.
func ConvertWith(subject string, data []byte, trees Trees, files Files, list Listed) (Message, error) {
var action string
switch subject {
case UpdatedSubject:
@@ -114,7 +126,7 @@ func Convert(subject string, data []byte, trees Trees, files Files) (Message, er
return Message{}, unreadable(subject, "no branch")
}
if subject == MergedSubject && p.Base == Trunk {
return merged(subject, p, trees, files)
return merged(subject, p, trees, files, list)
}
f := delivery.Forge{Kind: delivery.ForgePull, Repository: p.Owner + "/" + p.Repo, Branch: p.Head, Number: p.Number,
Head: p.HeadSHA, Action: action}
@@ -227,7 +239,7 @@ func inputsOf(owner, repo, ref string, dirs []string, files Files) (manifests ma
}
// merged is the merged message of a merge into the trunk: the trees of the merge commit, asked once.
func merged(subject string, p pull, trees Trees, files Files) (Message, error) {
func merged(subject string, p pull, trees Trees, files Files, list Listed) (Message, error) {
if !sha40.MatchString(p.MergeSHA) {
return Message{}, unreadable(subject, "the merge commit %q is not 40 hex digits", p.MergeSHA)
}
@@ -252,8 +264,15 @@ func merged(subject string, p pull, trees Trees, files Files) (Message, error) {
if err != nil {
return Message{}, unreadable(subject, "%v", err)
}
body, err := delivery.EncodeForge(delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number,
Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks})
f := delivery.Forge{Kind: delivery.ForgeMerged, Repository: repository, Branch: p.Head, Number: p.Number,
Head: p.HeadSHA, Merge: p.MergeSHA, Root: root, Trees: dirs, Manifests: manifests, Locks: locks}
if list != nil && delivery.SameRepository(repository, delivery.CatalogueRepository) {
if f.Placements, err = placementsOf(p.Owner, p.Repo, p.MergeSHA, list, files); err != nil {
return Message{}, fmt.Errorf("%w: %s: the placement files of %s at the merge commit %s: %v", ErrRetry, subject, repository, p.MergeSHA, err)
}
f.PlacementsRead = true
}
body, err := delivery.EncodeForge(f)
if err != nil {
return Message{}, unreadable(subject, "%v", err)
}
@@ -350,3 +369,30 @@ func OpenMessage(answer []byte) (Message, error) {
}
return Message{Event: delivery.OpenEvent(), Data: body}, nil
}
// placementsOf is every placement file of the catalogue at a commit, by node, its bytes exact: each `<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
}
modules/mesh-delivery-forge-bridge/internal/bridge/files_test.go
+41
View File
@@ -79,6 +79,47 @@ func TestMergedCarriesManifestsAndLocks(t *testing.T) { // 1448/AC1
if want := map[string]map[string]string{"modules/a": {"go.sum": hexSum(sum)}}; !reflect.DeepEqual(f.Locks, want) {
t.Errorf("locks = %v, want %v", f.Locks, want)
}
if f.Placements != nil || f.PlacementsRead {
t.Errorf("a merge of another repository than the catalogue carries placement: %v", f.Placements)
}
// Issue 1126 (ADR 0549 §2): a merge of the catalogue carries every placement file of its merge commit, by node; a file
// whose name is no node is left out; a listing or a file the forge cannot give is a retry, never a part.
catalogue := mergedBody(map[string]any{"repo": "mesh-catalog"})
forge.have["placement/ace.json"] = []byte(`{"v":1,"node":"ace","modules":[]}`)
forge.have["placement/g14.json"] = []byte(`{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`)
listing := map[string]string{"ace.json": "b1", "g14.json": "b2", "README.md": "b3", "a.b.json": "b4"}
var listedAt []string
list := func(owner, repo, sha, dir string) (map[string]string, error) {
listedAt = append(listedAt, owner+"/"+repo+"@"+sha+":"+dir)
return listing, nil
}
msg, err = ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list)
if err != nil {
t.Fatal(err)
}
if f, err = delivery.DecodeForge(msg.Data); err != nil {
t.Fatal(err)
}
want := map[string]string{"ace": `{"v":1,"node":"ace","modules":[]}`, "g14": `{"v":1,"node":"g14","modules":[{"module":"zsh"}]}`}
if !f.PlacementsRead || !reflect.DeepEqual(f.Placements, want) || !reflect.DeepEqual(listedAt, []string{"novox/mesh-catalog@" + merge + ":placement"}) {
t.Errorf("placements = %v (read %v), listed at %v", f.Placements, f.PlacementsRead, listedAt)
}
listing = map[string]string{}
if msg, err = ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list); err != nil {
t.Fatal(err)
}
if f, _ = delivery.DecodeForge(msg.Data); !f.PlacementsRead || f.Placements != nil {
t.Errorf("a catalogue with no placement file says it read none: %v, %v", f.Placements, f.PlacementsRead)
}
listing = map[string]string{"gone.json": "b5"}
if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, list); !errors.Is(err, ErrRetry) || !strings.Contains(err.Error(), "placement/gone.json") {
t.Errorf("a placement file the forge cannot give: %v", err)
}
failing := func(_, _, _, _ string) (map[string]string, error) { return nil, errors.New("forge down") }
if _, err := ConvertWith(mergedSubject, catalogue, treesOf(&calls, "r0", map[string]string{"modules/a": "t1"}, nil), forge.get, failing); !errors.Is(err, ErrRetry) {
t.Errorf("a listing the forge cannot give: %v", err)
}
}
func TestPullWithKnownDirsCarriesTheSameAtTheHead(t *testing.T) { // 1448/AC2
modules/mesh-delivery-indexer/cmd/mesh-delivery-indexer/main.go
+57 -1
View File
@@ -91,6 +91,58 @@ func pinned(want string) *tls.Config {
}
}
// awaitedStore is a state bound when it appears (process.AwaitBucket) as the indexer's store: until then every call is
// process.ErrNotYet.
type awaitedStore struct{ b *process.AwaitedBucket }
func (s awaitedStore) store() (kvStore, error) {
kv := s.b.KV()
if kv == nil {
return kvStore{}, process.ErrNotYet
}
return kvStore{kv}, nil
}
func (s awaitedStore) Get(ctx context.Context, key string) ([]byte, uint64, bool, error) {
st, err := s.store()
if err != nil {
return nil, 0, false, err
}
return st.Get(ctx, key)
}
func (s awaitedStore) Create(ctx context.Context, key string, value []byte) error {
st, err := s.store()
if err != nil {
return err
}
return st.Create(ctx, key, value)
}
func (s awaitedStore) Update(ctx context.Context, key string, value []byte, revision uint64) error {
st, err := s.store()
if err != nil {
return err
}
return st.Update(ctx, key, value, revision)
}
func (s awaitedStore) Delete(ctx context.Context, key string, revision uint64) error {
st, err := s.store()
if err != nil {
return err
}
return st.Delete(ctx, key, revision)
}
func (s awaitedStore) Keys(ctx context.Context) ([]string, error) {
st, err := s.store()
if err != nil {
return nil, err
}
return st.Keys(ctx)
}
// kvStore is the candidates bucket as the indexer's store: compare-and-set through Create, Update and Delete at a revision.
type kvStore struct{ kv jetstream.KeyValue }
@@ -257,6 +309,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
if err != nil {
return err
}
// Its placement state (ADR 0549 §2) is bound when it appears, as every state a delivery adds (issue 1790): until then a
// catalogue merge's placement is said and not written, and the rest of the indexer runs.
placement := process.AwaitBucket(ctx, js, conn, delivery.PlacementBucket, "no catalogue merge's placement is written", logger.Printf)
// The controller's build records only name the directory of a module whose key names none (fillPaths): without them
// such a module waits for its path, every other is indexed (issue 1945).
builds := process.OptionalBucket(ctx, js, delivery.BuildsBucket, "a module whose key names no directory is not given one from the controller's build records until the indexer starts again", logger.Printf)
@@ -291,7 +346,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
}
defer calls.Close()
ix := &index.Indexer{Store: kvStore{candidates}, Modules: kvStore{modules}, Manifests: kvStore{manifests}, Taken: kvStore{taken}, Builds: known, Logf: logger.Printf,
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call)}
Named: namedVia(calls.Call), Trunk: trunkVia(calls.Call), Contains: containsVia(calls.Call), Read: readVia(calls.Call),
Placement: awaitedStore{placement}, Say: func(event string, body any) error { return calls.Emit(event, body) }}
// The gate's source bucket (mesh-delivery-gate_source) comes with the gate (release 3e, novox/hq ADR 0450 §1): its key
// main is the mesh's source revision, which stamps the modules (issue 1344). Where it does not exist yet the indexer does
// not wait for it (that needs a timer, ADR 0437 §8): it says so, and is started again with the gate.
modules/mesh-delivery-indexer/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-indexer
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210442-b4310dfb8ec6
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-indexer/go.sum
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210442-b4310dfb8ec6 h1:JDgD6KA80fzQzVj/NFQD7hXVsZQfsYW6xjZp0MoPDUU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210442-b4310dfb8ec6/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-delivery-indexer/internal/index/candidates.go
+7
View File
@@ -65,6 +65,10 @@ type Indexer struct {
// Contains says whether a commit is the second or an ancestor of it (gitea.gitea_contains).
Contains func(ctx context.Context, repository, commit, in string) (bool, error)
Logf func(format string, args ...any)
// Placement is the placement bucket (delivery.PlacementBucket, ADR 0549 §2); nil writes no placement.
Placement Store
// Say says one of the indexer's events (placement-changed); nil says none.
Say func(event string, body any) error
mu sync.Mutex // guards source, toolchains and previews, which the watches and the forge messages all reach
source sourceNow
@@ -104,6 +108,9 @@ func (ix *Indexer) Apply(ctx context.Context, f delivery.Forge) error {
}
ix.kickRetry(ctx) // a registered module the forge could not read is read again on a merge, in the background (issue 1945)
}
if err := ix.applyPlacement(ctx, f); err != nil {
return err
}
var digests map[string]string
var held map[string]delivery.Module
if f.Kind == delivery.ForgePull {
modules/mesh-delivery-indexer/internal/index/manifests_test.go
+2
View File
@@ -43,6 +43,8 @@ func withManifests(f delivery.Forge, manifests map[string]string, locks map[stri
}
func TestAMergedManifestIsKeptByItsDigestAndNamedByTheModule(t *testing.T) { // 1449/AC1
// A merge's placement files are written as its manifests are (issue 1126): folded here to hold the count.
placementIsWrittenAtAMergeOnlyWhereItChanged(t)
avahi := catalogueManifest(t, "avahi")
ix, mods, manifests := manifestsIndexer(t, rec(t, "build-1", "avahi", catalogueRepo, "modules/avahi", "main", "", "2026-10-19T09:00:00Z", aInputs))
f := withManifests(mergedMsg(catalogueRepo, map[string]string{"modules/avahi": "newtree"}), map[string]string{"modules/avahi": avahi}, nil)
modules/mesh-delivery-indexer/internal/index/placement.go
+123
View File
@@ -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
}
modules/mesh-delivery-indexer/internal/index/placement_test.go
+97
View File
@@ -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)
}
}
modules/mesh-delivery-indexer/internal/index/previews_test.go
+2 -1
View File
@@ -164,6 +164,7 @@ func TestAPreviewHeardBeforeItsPullRequestGivesTheFingerprintWhenTheCandidateIsW
}
func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T) {
t.Run("the same preview again", testTheSamePreviewAppliedAgainWritesNothing)
ix, _, cands := previewBed(t, true)
if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil {
t.Fatal(err)
@@ -176,7 +177,7 @@ func TestAPreviewNamingNoManifestOfTheModulesDirectoryGivesAReason(t *testing.T)
}
}
func TestTheSamePreviewAppliedAgainWritesNothing(t *testing.T) {
func testTheSamePreviewAppliedAgainWritesNothing(t *testing.T) {
ix, _, cands := previewBed(t, true)
if err := ix.Apply(context.Background(), avahiPull(t, h1)); err != nil {
t.Fatal(err)
modules/mesh-delivery-indexer/manifest_test.go
+8 -3
View File
@@ -38,6 +38,7 @@ type manifest struct {
Version string `json:"version"`
Slug string `json:"slug"`
Tools []string `json:"tools"`
Emits []string `json:"emits"`
Consumes []string `json:"consumes"`
State []state `json:"state"`
Reads []string `json:"reads"`
@@ -80,7 +81,11 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
if m.Module != "mesh-delivery-indexer" {
t.Errorf("module = %q", m.Module)
}
if want := []state{{Name: "candidates", History: 10}, {Name: "manifests", History: 1}, {Name: "modules", History: 20}, {Name: "taken", History: 1}, {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: "taken", History: 1}, {Name: "takeins", History: 1}}; !reflect.DeepEqual(m.State, want) {
t.Errorf("state = %+v, want %+v", m.State, want)
}
if want := []string{"mesh-build-go.toolchains", "mesh-build-node.toolchains", "mesh-controller.assignments", "mesh-controller.builds", "mesh-delivery-checker.previews", "mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
@@ -107,8 +112,8 @@ func TestManifestDeclarations(t *testing.T) { // 1238/AC1, 1444 (reads the build
if want := []string{"indexer_take_in"}; !reflect.DeepEqual(m.Tools, want) {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
// No emits, no streams, no bus, no seat (ADR 0437 §4).
for _, k := range []string{"streams", "logs", "emits", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
// No streams, no bus, no seat (ADR 0437 §4); its one event is placement-changed (ADR 0549 §2), checked above.
for _, k := range []string{"streams", "logs", "claims", "provides", "requires", "uses", "defines-seats", "seat", "seats", "bus"} {
if _, ok := top[k]; ok {
t.Errorf("the manifest declares %q: the indexer has none", k)
}
modules/mesh-delivery-indexer/module.json
+4
View File
@@ -5,6 +5,9 @@
"tools": [
"indexer_take_in"
],
"emits": [
"placement-changed"
],
"consumes": [
"mesh-build-go.toolchain.reported",
"mesh-build-node.toolchain.reported",
@@ -15,6 +18,7 @@
{ "name": "candidates", "history": 10 },
{ "name": "manifests", "history": 1 },
{ "name": "modules", "history": 20 },
{ "name": "placement", "history": 10 },
{ "name": "taken", "history": 1 },
{ "name": "takeins", "history": 1 }
],
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main.go
+11 -1
View File
@@ -62,6 +62,15 @@ func Tools(r view.Reader, u view.UsageReader) []stdio.Tool {
mods, err := r.Modules(only)
return map[string]any{"modules": mods}, err
}},
{Name: view.PlacementReadTool,
Description: "Where modules run, as the indexer read it from the catalogue's placement files at the last merge (placement.<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,
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 " +
@@ -193,7 +202,8 @@ func read(state stdio.KeptState) view.StateRead {
func main() {
r := view.Reader{State: read(stdio.State("mesh-delivery-indexer.modules")), Builds: read(stdio.State("mesh-delivery-builder.builds")), CandidatesState: read(stdio.State("mesh-delivery-indexer.candidates")),
WantsState: read(stdio.State("mesh-delivery-assigner.wants")), DifferencesState: read(stdio.State("mesh-delivery-assigner.differences")), HavesState: read(stdio.State("node-engine.have")),
ReachState: read(stdio.State("mesh-delivery-view.reach"))}
ReachState: read(stdio.State("mesh-delivery-view.reach")),
PlacementState: read(stdio.State("mesh-delivery-indexer.placement"))}
u := view.UsageReader{Haves: r.HavesState, Resolved: read(stdio.State("mesh-config.resolved")), Tries: read(stdio.State("mesh-delivery-checker.tries")),
Steps: map[string]view.StateRead{}, Pauses: map[string]view.StateRead{}, Backlog: backlogOf(socket())}
for _, lang := range delivery.AskLanguages {
modules/mesh-delivery-view/cmd/mesh-delivery-view-tools/main_test.go
+26
View File
@@ -105,6 +105,32 @@ func TestModulesToolAnswersEveryKeyAsStored(t *testing.T) {
if len(only) != 1 || string(only["b"]) != withNone {
t.Errorf("none2 modules = %v", only)
}
// 1126 (ADR 0549 §2): indexer_placement answers a node's key as stored, every node placing a module, or every node's,
// and lists a key that does not read.
const ace, g14 = `{"v":1,"node":"ace","revision":"m1","modules":[{"module":"plex"}]}`, `{"v":1,"node":"g14","revision":"m1","modules":[{"module":"zsh"},{"module":"plex"}]}`
p := view.Reader{PlacementState: fake{"placement.ace": ace, "placement.g14": g14, "placement.bad": `{"v":1}`, "other": "{}"}.reader().State}
got := func(args map[string]any) map[string]json.RawMessage {
t.Helper()
out, err := run(t, p, view.PlacementReadTool, args)
if err != nil {
t.Fatal(err)
}
return out.(map[string]any)["placement"].(map[string]json.RawMessage)
}
if one := got(map[string]any{"node": "ace"}); len(one) != 1 || string(one["ace"]) != ace {
t.Errorf("node ace = %v", one)
}
if zsh := got(map[string]any{"module": "zsh"}); len(zsh) != 1 || string(zsh["g14"]) != g14 {
t.Errorf("module zsh = %v", zsh)
}
out, err = run(t, p, view.PlacementReadTool, map[string]any{})
if all := out.(map[string]any); err != nil || len(all["placement"].(map[string]json.RawMessage)) != 2 || len(all["unreadable"].([]string)) != 2 {
t.Errorf("every node = %v, %v", all, err)
}
if _, err := run(t, p, view.PlacementReadTool, map[string]any{"node": "novox"}); err == nil || !strings.Contains(err.Error(), "no placement of novox") {
t.Errorf("a node with no key: %v", err)
}
}
const (
modules/mesh-delivery-view/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-delivery-view
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210442-b4310dfb8ec6
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61
github.com/nats-io/nats-server/v2 v2.11.17
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-delivery-view/go.sum
+2 -2
View File
@@ -1,5 +1,5 @@
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210442-b4310dfb8ec6 h1:JDgD6KA80fzQzVj/NFQD7hXVsZQfsYW6xjZp0MoPDUU=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210442-b4310dfb8ec6/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61 h1:/c6ncICWjq55nBf8TlDM0ROe/PuH+qZIdnuVbtg8UuM=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026211913-68d65e6bbf61/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-delivery-view/internal/view/read.go
+55
View File
@@ -20,6 +20,8 @@ import (
const (
ModuleReadTool = "indexer_module"
ModulesReadTool = "indexer_modules"
// PlacementReadTool reads the indexer's placement (novox/hq ADR 0549 §2, issue 1126).
PlacementReadTool = "indexer_placement"
)
// StateRead reads the indexer's modules state, as the view's reads grant it: its keys and a key's stored value (nil for a key with none).
@@ -40,6 +42,7 @@ type Reader struct {
DifferencesState StateRead // the assigner's differences
HavesState StateRead // the node-engine's haves
ReachState StateRead // the view's own reach (ADR 0556)
PlacementState StateRead // the indexer's placement
}
// ReachReadTool reads the view's reach: per repository, how far each module reached and the reached head (ADR 0556).
@@ -645,3 +648,55 @@ func (r Reader) Differences(node, module string) (KeptAnswer, error) {
func (r Reader) Haves(node, module string) (KeptAnswer, error) {
return r.kept(r.HavesState, "the haves", node, module)
}
// Placement is the indexer's placement as stored (ADR 0549 §2): one node's key with node, every node placing a module
// with module (each node's key whole, so its pins, ports and bindings are beside the module), and every node's key with
// neither. A key that does not read as a placement is listed as unreadable, never dropped.
func (r Reader) Placement(node, module string) (map[string]any, error) {
node, module = strings.TrimSpace(node), strings.TrimSpace(module)
if r.PlacementState.Keys == nil || r.PlacementState.Get == nil {
return nil, fmt.Errorf("the placement state cannot be read here")
}
if node != "" {
key, err := delivery.PlacementKey(node)
if err != nil {
return nil, err
}
raw, err := r.PlacementState.Get(key)
if err != nil {
return nil, err
}
if len(raw) == 0 {
return nil, fmt.Errorf("no placement of %s: the catalogue holds no placement file for it, or the indexer has read none yet", node)
}
return map[string]any{"placement": map[string]json.RawMessage{node: raw}}, nil
}
keys, err := r.PlacementState.Keys()
if err != nil {
return nil, err
}
sort.Strings(keys)
values, err := r.PlacementState.values(keys)
if err != nil {
return nil, err
}
out := map[string]json.RawMessage{}
var unreadable []string
for i, key := range keys {
n, err := delivery.ParsePlacementKey(key)
if err != nil {
unreadable = append(unreadable, key)
continue
}
p, err := delivery.DecodePlacement(values[i])
if err != nil {
unreadable = append(unreadable, key)
continue
}
if module != "" && !slices.Contains(p.Assigned(), module) {
continue
}
out[n] = values[i]
}
return map[string]any{"placement": out, "unreadable": unreadable}, nil
}
modules/mesh-delivery-view/manifest_test.go
+2 -2
View File
@@ -84,7 +84,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
if m.Module != "mesh-delivery-view" {
t.Errorf("module = %q", m.Module)
}
want := []string{"mesh-build-go.pauses", "mesh-build-go.steps", "mesh-build-image.pauses", "mesh-build-image.steps", "mesh-build-node.pauses", "mesh-build-node.steps", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-delivery-assigner.asks", "mesh-delivery-assigner.differences", "mesh-delivery-assigner.wants", "mesh-delivery-builder.builds", "mesh-delivery-checker.tries", "mesh-delivery-gate.asks", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.modules", "mesh-delivery-indexer.taken", "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-indexer.taken", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-delivery-watchman.stuck", "node-engine.have"} // build_usage reads the build modules' steps and pauses, mesh-config's capacities and the checker's tries; 1168/AC1, 1241/AC4, 1345/AC3, 1297/B8, B9
got := slices.Clone(m.Reads)
slices.Sort(got)
if !reflect.DeepEqual(got, want) {
@@ -116,7 +116,7 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
}
// The two read tools of the indexer's modules (issue 1562), carried by a bundle the runtime launches; the indexer
// itself serves no tool (ADR 0437 §4). Carrying them gives the tool runner read access to every bucket the view reads.
if want := []string{"indexer_module", "indexer_modules", "indexer_candidate", "indexer_candidates", "builder_build", "builder_builds", "assigner_wants", "assigner_want", "assigner_differences", "node_haves", "build_usage", "view_reach"}; !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", "view_reach"}; !reflect.DeepEqual(m.Tools, want) {
t.Errorf("tools = %v, want %v", m.Tools, want)
}
for _, a := range m.Build.Artifacts {
modules/mesh-delivery-view/module.json
+2
View File
@@ -6,6 +6,7 @@
"tools": [
"indexer_module",
"indexer_modules",
"indexer_placement",
"indexer_candidate",
"indexer_candidates",
"builder_build",
@@ -47,6 +48,7 @@
"mesh-delivery-gate.source",
"mesh-delivery-indexer.candidates",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.placement",
"mesh-delivery-indexer.taken",
"mesh-delivery-teller.answers",
"mesh-delivery-watchman.presence",
modules/mesh-grants/go.mod
+1 -1
View File
@@ -3,7 +3,7 @@ module mesh-grants
go 1.26.0
require (
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026142434-65ee43f07583
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117
github.com/nats-io/nats.go v1.54.0
)
modules/mesh-grants/go.sum
+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.20261026142434-65ee43f07583/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117 h1:88CNZOoQgu7YPYKW16qkWT8Gw9Ucg07aTrIImA2nYUk=
git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261026210326-979644420117/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op h1:Z/MZK75wC/NSrkgqeNIa7jexam9uWzhLmFTSCPI/kn0=
github.com/antithesishq/antithesis-sdk-go v0.7.0-default-no-op/go.mod h1:FQyySiasQQM8735Ddel3MRojmy4dA1IqCeyJ5jmPMbI=
github.com/google/go-tpm v0.9.8 h1:slArAR9Ft+1ybZu0lBwpSmpwhRXaa85hWtMinMyRAWo=
modules/mesh-grants/grants_test.go
+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) {
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"}}
modules/mesh-grants/internal/grants/follow.go
+1 -1
View File
@@ -18,7 +18,7 @@ type Watcher interface {
// Followed are the states whose change composes again: placement, the indexer's manifests by module, the resolved
// config values, the node agents' haves (a machine's first have, and the bus node's loaded list) and the assigner's
// wants (the manifest each machine is wanted to run), and the vault's hashes of the bus accounts.
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts}
var Followed = []string{ReadAssignments, ReadModules, ReadResolved, ReadHave, ReadWants, ReadAccounts, ReadPlacement}
// Follower composes again on every change it hears, never on a timer. It never stops for a refusal: a watch it cannot
// open is said, composed without, and opened again at the next wake (any change it does hear, or a reconnect).
modules/mesh-grants/internal/grants/inputs.go
+41 -2
View File
@@ -34,11 +34,14 @@ const (
ReadBuilds = "mesh-delivery-builder.builds"
// ReadAccounts is the vault's bcrypt hashes of the bus accounts it holds, by bus user (ADR 0547 clause 9).
ReadAccounts = "mesh-vault.bus-accounts"
// ReadPlacement is the indexer's placement, read from the catalogue (ADR 0549 §2, §3): each node's modules and the seats
// they hold.
ReadPlacement = "mesh-delivery-indexer.placement"
)
// Reads is every state the part reads, in the manifest's order.
var Reads = []string{ReadAssignments, ReadControllerGrant, ReadModules, ReadManifests, ReadResolved, ReadHave, ReadWants, ReadBuilds,
ReadAccounts}
ReadAccounts, ReadPlacement}
// The bus module's tools the part calls, as its manifest's invokes name them: read-only, each.
const (
@@ -103,8 +106,12 @@ type resolved struct {
type Inputs struct {
// Nodes is every machine seen: in a node agent's have or a standing assignment.
Nodes []string
// Assigned is each machine's standing assignments, sorted.
// Assigned is each machine's standing assignments and what the catalogue's placement puts there, sorted: while both
// stand (ADR 0549 §5) a module either names keeps its membership.
Assigned map[string][]string
// Placed is each node's placement as the indexer read it from the catalogue (ADR 0549 §2); empty before any is read.
// Where it is, a seat held once for the mesh is held where placement says, not wherever a module claiming it runs.
Placed map[string]delivery.Placement
// Manifests is each module's manifest at the indexer's trunk, by module.
Manifests map[string]catalogue.Manifest
// Digests is each module's manifest digest, by module.
@@ -230,6 +237,38 @@ func ReadInputs(r Reader, cache *Manifests) Inputs {
in.BusNode = a.Node
}
}
// The catalogue's placement (ADR 0549 §3): what it places is composed beside the assignments, and the seats held once
// for the mesh are held where it says. Not on the bus yet is a standing gap, never a failed read, so the list is still
// composed and sealed from the assignments alone.
if keys, err := r.Keys(ReadPlacement); err != nil {
in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement, Why: "not read yet: the assignments alone place modules (" + err.Error() + ")"})
} else {
for _, k := range keys {
raw, found, err := r.Get(ReadPlacement, k)
if err != nil || !found {
continue
}
p, err := delivery.DecodePlacement(raw)
node, kerr := delivery.ParsePlacementKey(k)
if err != nil || kerr != nil || p.Node != node {
in.Gaps = append(in.Gaps, Gap{Input: ReadPlacement + " " + k, Why: "not a placement, so the node is composed from its assignments alone"})
continue
}
if in.Placed == nil {
in.Placed = map[string]delivery.Placement{}
}
in.Placed[node] = p
nodes[node] = true
for _, m := range p.Assigned() {
if !slices.Contains(in.Assigned[node], m) {
in.Assigned[node] = append(in.Assigned[node], m)
}
}
}
}
if len(in.Placed) > 0 {
in.Gaps = slices.DeleteFunc(in.Gaps, func(g Gap) bool { return g.Input == "seat holdings" })
}
haveKeys, err := r.Keys(ReadHave)
if err != nil {
in.Gaps = append(in.Gaps, Gap{Input: ReadHave, Why: "refused or absent: " + err.Error(), Failed: true})
modules/mesh-grants/internal/grants/records.go
+23
View File
@@ -7,6 +7,7 @@ import (
"git.novox.be/novox/mesh-sdk/go/buscompose"
"git.novox.be/novox/mesh-sdk/go/catalogue"
"git.novox.be/novox/mesh-sdk/go/delivery"
)
// Records is what the bus composer composes from, rebuilt from the bus (mesh-controller internal/inventory busrecords.go,
@@ -59,6 +60,9 @@ func Records(in Inputs) (buscompose.Records, []Gap) {
continue
}
d := declaredFor(m, seats, declarers)
if len(in.Placed) > 0 {
d.Holds = heldWherePlaced(d.Holds, in.Placed[node], m.Module)
}
for _, sr := range catalogue.ReadsGranted(m, here, node) {
d.KeyedReads = append(d.KeyedReads, buscompose.KeyedRead{Read: sr.Read, Key: sr.Key})
}
@@ -299,3 +303,22 @@ func hearsOf(m catalogue.Manifest, declared map[string]catalogue.Manifest, node
}
return out, touches, onEvents
}
// heldWherePlaced is a module's holds on a node as the catalogue's placement says them (ADR 0549 §1, §3): a seat held once
// for the mesh is held there only when the node's placement names this module holding it, so a module claiming such a seat
// on two machines holds it on the one placement names, not on both; a seat of every node is held wherever the module runs.
func heldWherePlaced(holds []buscompose.Seat, p delivery.Placement, module string) []buscompose.Seat {
var out []buscompose.Seat
for _, h := range holds {
if h.Scope != "" && h.Scope != catalogue.ScopeMesh {
out = append(out, h)
continue
}
if slices.ContainsFunc(p.Holds, func(x delivery.Holding) bool {
return x.Claim == h.Name && x.Module == module && x.Kind == h.Kind
}) {
out = append(out, h)
}
}
return out
}
modules/mesh-grants/module.json
+1
View File
@@ -10,6 +10,7 @@
"mesh-delivery-builder.builds",
"mesh-delivery-indexer.manifests",
"mesh-delivery-indexer.modules",
"mesh-delivery-indexer.placement",
"node-engine.have",
"mesh-vault.bus-accounts"
],