Assigner: read the running commit of a reader of the manifests from the builder's log, and hold the buckets it starts without in one list so the start test skips them (issue 1955, ADR 0571 §1)

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/asks.go

@@ -119,11 +119,11 @@ func (p *passer) composeAtHead(ctx context.Context, m assign.Mesh, ch assign.Cha
 		if h := ch.Members[repository]; h != "" {
 			commits, member = []string{h}, true
 		} else {
-			commits = p.w.runningCommits(ctx, kvGetter(p.kvs[delivery.BuildsBucket]), module)
+			commits = p.w.runningCommits(module)
 		}
 		if len(commits) == 0 {
-			notJudged = append(notJudged, "not judged whether "+module+" reads the group's manifests: no build of it is assigned with a "+
-				"commit on record, and the group holds no member of "+repository)
+			notJudged = append(notJudged, "not judged whether "+module+" reads the group's manifests: the builder's log holds no done "+
+				"trunk build of it, and the group holds no member of "+repository)
 			continue
 		}
 		for _, commit := range commits {

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/compose_integration_test.go

@@ -106,7 +106,7 @@ func TestAGroupWithAnSDKMemberIsComposedByTheComposerAtItsHead(t *testing.T) {
 	b.putJSON(t, provider.AssignmentsBucket, "g14.web", []byte(`{"node":"g14","module":"web","build":"build-1"}`))
 	running := strings.Repeat("3", 40)
 	b.putJSON(t, provider.AssignmentsBucket, "novox.mesh-controller", []byte(`{"node":"novox","module":"mesh-controller","build":"c-1"}`))
-	b.putJSON(t, delivery.BuildsBucket, delivery.BuildKey("mesh-controller", "c-1"), []byte(`{"id":"c-1","repository":"novox/mesh-controller","commit":"`+running+`"}`))
+	b.built(t, "mesh-controller", "novox/mesh-controller", running)
 	sdkHead := strings.Repeat("4", 40)
 	heads := []delivery.Head{{Repository: "novox/mesh-sdk", Number: 80, Head: sdkHead}, {Repository: "novox/web", Number: 7, Head: strings.Repeat("2", 40)}}
 	hash, _ := delivery.HeadsHash(heads)

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/integration_test.go

@@ -21,6 +21,7 @@ import (
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/announce"
 	"git.novox.be/novox/mesh-sdk/go/asks"
+	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/compose/judge"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
@@ -82,6 +83,11 @@ func startBed(t *testing.T, leaveOut string) *bed {
 		}
 		b.buckets[name] = kv
 	}
+	// The builder's module log, as the runtime raises it (ADR 0297): the running commits are read from it (ADR 0571 clause 1).
+	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: buscompose.LogStreamName(delivery.BuilderModule, delivery.BuilderLog),
+		Subjects: []string{buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog) + ".>"}, AllowDirect: true}); err != nil {
+		t.Fatal(err)
+	}
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "EVENTS", Subjects: []string{"mesh.mod.*.event.>"}}); err != nil {
 		t.Fatal(err)
 	}
@@ -177,8 +183,8 @@ func saysWhatItKeeps(t *testing.T, b *bed, done <-chan error) {
 // 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 || sourceOfGrants(missing) {
-			continue // it starts without it, below (the grants' source: main_test.go)
+		if startsWithout(missing) {
+			continue // it starts without it (the placement and the grants' source: main_test.go; the puts: below)
 		}
 		b := startBed(t, missing)
 		select {
@@ -264,6 +270,19 @@ func (b *bed) putJSON(t *testing.T, bucket, key string, value []byte) {
 	}
 }
 
+// built appends the end of a done trunk build of module at commit to the builder's log, as the builder does.
+func (b *bed) built(t *testing.T, module, repository, commit string) {
+	t.Helper()
+	raw, err := delivery.EncodeBuildLogEntry(delivery.BuildLogEntry{Event: delivery.BuildEndedEvent, Module: module, FP: delivery.SourcePrefix2 + strings.Repeat("c", 64),
+		Repository: repository, Directory: delivery.RootPath, Commit: commit, Node: "novox", At: time.Now().UTC(), Outcome: delivery.BuildDone})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := b.js.Publish(b.ctx, buscompose.LogSubject(delivery.BuilderModule, delivery.BuilderLog)+"."+module, raw); err != nil {
+		t.Fatal(err)
+	}
+}
+
 // compose says the controller composed module for node at fp, with the digest d (a letter repeated) and the judgement of the
 // move to it, in mesh-controller_composed.
 func (b *bed) compose(t *testing.T, node, module, fp string, d byte, judgement delivery.Judgement) string {

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main.go

@@ -22,6 +22,7 @@ import (
 	"github.com/nats-io/nats.go"
 	"github.com/nats-io/nats.go/jetstream"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/announce"
 	"git.novox.be/novox/mesh-sdk/go/asks"
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
@@ -58,7 +59,6 @@ func main() {
 var readBuckets = []string{
 	buscompose.ResolvedBucket,
 	provider.AssignmentsBucket,
-	delivery.BuildsBucket,
 	delivery.ControllerGrantsBucket,
 	delivery.PartGrantsBucket,
 	buscompose.LeaseBucket,
@@ -76,6 +76,10 @@ var readBuckets = []string{
 	keptBucket,
 }
 
+// readLogs are the logs of other parts the assigner reads, each named in its manifest's reads: the builder's log, whose
+// newest done trunk build of a module names the commit a reader of the manifests runs at (novox/hq ADR 0571 clause 1).
+var readLogs = []string{delivery.BuildLogRead}
+
 // takeoverBucket is the condition keeper's take-over key's bucket (novox/hq ADR 0564 §9): whether the assigner's said
 // conditions go straight to the router or only as findings (issue 1952 WM6).
 var takeoverBucket = health.Bucket(health.KeeperModule, health.TakeoverState)
@@ -204,6 +208,17 @@ func open(ctx context.Context, js process.BucketOpener, logf func(string, ...any
 	return out, nil
 }
 
+// startsWithout says the assigner starts without the bucket when it cannot be opened (open says why for each): the seat's
+// states and those new with a delivery of their own, bound once made, and the two that choose the grants' source.
+func startsWithout(name string) bool {
+	switch name {
+	case delivery.SeatHoldersBucket, keptBucket, provisions.ProvisionsBucket, assign.PoliciesBucket, assign.FindingsBucket, takeoverBucket,
+		delivery.AssignerPutsBucket:
+		return true
+	}
+	return slices.Contains(seatBuckets, name) || sourceOfGrants(name)
+}
+
 // sourceOfGrants says the bucket is one of the two that choose the grants' source, the grants part's grants and the
 // controller's lease: neither stops the assigner when it is not there (delivery.GrantSource).
 func sourceOfGrants(name string) bool {
@@ -239,7 +254,10 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	thisNode = account.Node
 	logger.Printf("started on the account %s, reading %d buckets", account.User, len(kvs))
 	findings := findingsOf(ctx, kvs, logger.Printf)
-	return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger)
+	buildLog := func(handle func(stdio.LogEntry) error) error {
+		return said.ReadLog(delivery.BuilderModule, delivery.BuilderLog).Watch("", stdio.LogFromAll, handle)
+	}
+	return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, buildLog, logger)
 }
 
 // findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, and the
@@ -341,7 +359,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, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, logger *log.Logger) error {
+func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, kvs buckets, say func(asks.Condition) error, findings assign.Findings, bus caller, buildLog logWatch, logger *log.Logger) error {
 	w := newWorld()
 	dirty := make(chan struct{}, 1)
 	wake := func() {
@@ -350,6 +368,22 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 		default:
 		}
 	}
+	// The builder's log (ADR 0571 clause 1): every entry it keeps before the first pass, then each new one as it is appended;
+	// a module's newer done trunk build makes a pass, as a composition waiting on its commit is judged then. Where it cannot be
+	// read, a composition names no running reader of the manifests until the assigner starts again, and it is said.
+	if buildLog != nil {
+		var first atomic.Bool
+		if err := buildLog(func(le stdio.LogEntry) error {
+			if w.putBuilt(le.Value) && first.Load() {
+				wake()
+			}
+			return nil
+		}); err != nil {
+			logger.Printf("%s is not read (%v): no composition is judged against a running reader of the manifests until the assigner starts again",
+				delivery.BuildLogRead, err)
+		}
+		first.Store(true)
+	}
 	errs := make(chan error, len(watched)+2)
 	settled := make(chan struct{}, len(watched)+2)
 	watch := func(name string, opts ...jetstream.WatchOpt) error {
@@ -433,7 +467,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 	}
 	started := 0
 	for _, name := range watched {
-		if kvs[name] == nil && (slices.Contains(seatBuckets, name) || sourceOfGrants(name) || name == provider.AssignmentsBucket || name == assign.PoliciesBucket || name == provisions.ProvisionsBucket || name == keptBucket || name == delivery.SeatHoldersBucket) {
+		if kvs[name] == nil && (startsWithout(name) || name == provider.AssignmentsBucket) {
 			continue // not on the bus at this start (open said so)
 		}
 		started++

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/main_test.go

@@ -42,8 +42,16 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
 	}
 	var reads []string
 	for _, r := range m.Reads {
+		if slices.Contains(readLogs, r) {
+			continue // a log, read through its own reader (ADR 0571 clause 1), not a bucket
+		}
 		reads = append(reads, strings.Replace(r, ".", "_", 1))
 	}
+	for _, l := range readLogs {
+		if !slices.Contains(m.Reads, l) {
+			t.Errorf("the program reads the log %s, the manifest does not name it", l)
+		}
+	}
 	var states []string
 	for _, s := range m.State {
 		states = append(states, m.Module+"_"+s.Name)
@@ -112,7 +120,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, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, log.New(io.Discard, "", 0))
+	go follow(ctx, nil, nil, kvs, func(asks.Condition) error { return nil }, assign.Findings{}, nil, nil, log.New(io.Discard, "", 0))
 	select {
 	case key := <-wrote:
 		if key != "want.ace.node" {

modules/mesh-delivery-assigner/cmd/mesh-delivery-assigner/world.go

@@ -13,6 +13,7 @@ import (
 
 	"github.com/nats-io/nats.go/jetstream"
 
+	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/compose"
@@ -51,6 +52,8 @@ type world struct {
 	changes map[string]uint64
 	// meta is, by bucket and key, the revision and the time of the value held now, where the watch said them (putEntry).
 	meta map[string]entryMeta
+	// trunkBuilt is, by module, its newest done trunk build in the builder's log (putBuilt).
+	trunkBuilt map[string]delivery.BuildLogEntry
 }
 
 // entryMeta is a value's revision in its bucket and when it was written.
@@ -68,7 +71,7 @@ type parsed struct {
 
 func newWorld() *world {
 	return &world{values: map[string]map[string][]byte{}, manifests: map[string]parsed{}, healthy: map[string][]string{}, changes: map[string]uint64{},
-		meta: map[string]entryMeta{}}
+		meta: map[string]entryMeta{}, trunkBuilt: map[string]delivery.BuildLogEntry{}}
 }
 
 // putEntry is put, keeping the value's revision and the time it was written.
@@ -666,32 +669,35 @@ func (w *world) changed(bucket, key string) uint64 {
 	return w.changes[bucket+"|"+key]
 }
 
-// runningCommits is, of every open assignment of module, the commit of the build it names, read from the controller's
-// build record (`mesh-controller_builds` `<module>.build.<id>`), distinct and sorted: the commits a reader of the
-// manifests runs at (novox/hq ADR 0482 clause 5, issue 1623). A build whose record names no full commit is left out.
-func (w *world) runningCommits(ctx context.Context, get getter, module string) []string {
-	_, all := w.assigned()
-	seen := map[string]bool{}
-	for _, a := range all {
-		if a.Module != module || a.Build == "" {
-			continue
-		}
-		value, err := get(ctx, delivery.BuildKey(module, a.Build))
-		if err != nil {
-			continue
-		}
-		var rec delivery.BuildRecord
-		if json.Unmarshal(value, &rec) != nil || len(rec.Commit) != 40 {
-			continue
-		}
-		seen[rec.Commit] = true
+// logWatch follows the builder's log: handle each entry it keeps, then each new one (process.LogReader.Watch).
+type logWatch func(handle func(stdio.LogEntry) error) error
+
+// putBuilt takes one entry of the builder's log in: a done trunk build's end (no ref) naming a full commit, newer than the
+// module's last, becomes the module's running commit. It answers whether the commit changed.
+func (w *world) putBuilt(raw []byte) bool {
+	e, err := delivery.DecodeBuildLogEntry(raw)
+	if err != nil || e.Event != delivery.BuildEndedEvent || e.Outcome != delivery.BuildDone || e.Ref != "" || len(e.Commit) != 40 {
+		return false
 	}
-	out := make([]string, 0, len(seen))
-	for c := range seen {
-		out = append(out, c)
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	have, ok := w.trunkBuilt[e.Module]
+	if ok && !e.At.After(have.At) {
+		return false
 	}
-	sort.Strings(out)
-	return out
+	w.trunkBuilt[e.Module] = e
+	return !ok || have.Commit != e.Commit
+}
+
+// runningCommits is the commit of module's newest done trunk build in the builder's log (ADR 0571 clause 1): the commit a
+// reader of the manifests runs at (novox/hq ADR 0482 clause 5, issue 1623); none when the log holds no such build.
+func (w *world) runningCommits(module string) []string {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	if e, ok := w.trunkBuilt[module]; ok {
+		return []string{e.Commit}
+	}
+	return nil
 }
 
 // keptBucket is mesh-config's `kept`: the approved held values of open pull requests (ADR 0565 §5).

modules/mesh-delivery-assigner/go.mod

@@ -3,7 +3,7 @@ module mesh-delivery-assigner
 go 1.26.0
 
 require (
-	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121
+	git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641
 	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

@@ -1,5 +1,5 @@
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121 h1:OsA1v6n736a7RqEV8Nkip8pf5ILMbzj0PWuKqcBOhK0=
-git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029220947-f9a496d3e121/go.mod h1:ouERNpZizzwZ0HLEtAypsY9FR20FyUgUYRni6J7kVCY=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641 h1:nP/jrjm2Qt1wdmd1Bc+E6c4+PobAzW5o64SPArE+jaw=
+git.novox.be/novox/mesh-sdk/go v0.1.14-0.20261029222821-bb993dd60641/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/manifest_test.go

@@ -33,8 +33,8 @@ func TestManifest(t *testing.T) {
 	}
 	reads := slices.Clone(m.Reads)
 	slices.Sort(reads)
-	want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.kept", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.builds", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease",
-		"mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
+	want := []string{"mesh-build-go.steps", "mesh-conditions.takeover", "mesh-config.kept", "mesh-config.resolved", "mesh-controller.assignments", "mesh-controller.composed", "mesh-controller.grants", "mesh-controller.lease",
+		"mesh-delivery-builder.build-log", "mesh-delivery-builder.builds", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests",
 		"mesh-delivery-indexer.modules", "mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-provisioner.provisions", "node-engine.have"}
 	if !reflect.DeepEqual(reads, want) {
 		t.Errorf("reads = %v, want exactly %v", reads, want)

modules/mesh-delivery-assigner/module.json

@@ -31,10 +31,10 @@
     "mesh-config.kept",
     "mesh-config.resolved",
     "mesh-controller.assignments",
-    "mesh-controller.builds",
     "mesh-controller.composed",
     "mesh-controller.grants",
     "mesh-controller.lease",
+    "mesh-delivery-builder.build-log",
     "mesh-delivery-builder.builds",
     "mesh-delivery-indexer.candidates",
     "mesh-delivery-indexer.manifests",