DLV-M1: the builder follows the indexer's modules, manifests and candidates and the assigner's placement by their read tools' first views

ADR 0615 §1: the builder invokes indexer_read_modules, _manifests, _candidates and
assigner_read_placement, consumes their views families and reads none of the four buckets; it keeps
first-views and findings. The lead's ruling: an owner that is down never holds the start; the modules
are built once the modules', manifests' and candidates' first views are answered, the placement is
followed into the toolchain gate once its first view is, and until then the gate passes every report on
as before the placement settled. Tools builder_first_views and builder_first_view_again.

Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig

modules/mesh-delivery-builder/cmd/mesh-delivery-builder-tools/main.go

@@ -17,6 +17,7 @@ import (
 	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun"
 	stdio "git.novox.be/novox/mesh-sdk/go"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/views"
 
 	"mesh-delivery-builder/internal/build"
 )
@@ -264,7 +265,11 @@ func main() {
 		a, err := build.DecodeReAsk(v)
 		return err == nil && !a.Answered.IsZero()
 	})
-	if err := stdio.Serve("", Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now)); err != nil {
+	// builder_first_views and builder_first_view_again: where the builder's first views of the indexer's and the assigner's
+	// states stand, and asking one again by hand (ADR 0615, DLV-M1).
+	tools := append(Tools(r, put, askQueue(stdio.State(build.QueuesState), 25*time.Second), buildAgain, stdio.Log(delivery.BuilderLog).Since, time.Now),
+		views.StatusTools(delivery.BuilderModule, "builder_first_views", "builder_first_view_again", stdio.State(views.StatusState), time.Now)...)
+	if err := stdio.Serve("", tools); err != nil {
 		fmt.Fprintf(os.Stderr, "[mesh-delivery-builder-tools] %v\n", err)
 		os.Exit(1)
 	}

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

@@ -27,6 +27,7 @@ import (
 	runner "git.novox.be/novox/mesh-catalog/modules/mesh-build-go/buildstep"
 	"git.novox.be/novox/mesh-sdk/go/buildstep"
 	"git.novox.be/novox/mesh-sdk/go/buscompose"
+	"git.novox.be/novox/mesh-delivery/sdk/firstview/firstviewtest"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/process"
 
@@ -270,6 +271,13 @@ func startBedWith(t *testing.T, closed bool) *bed {
 	b.candidates = mk(delivery.CandidatesBucket, delivery.CandidatesHistory)
 	b.source = mk(delivery.SourceBucket, delivery.SourceHistory)
 	mk(delivery.ManifestsBucket, delivery.ManifestsHistory)
+	// The indexer and the assigner as the owners of the states the builder follows: their read tools and views families
+	// (ADR 0615, DLV-M1); n1 and n2 each have their durable consumer.
+	readers := []string{"n1_" + delivery.BuilderModule, "n2_" + delivery.BuilderModule}
+	for _, st := range []string{modulesState, manifestsState, candidatesState} {
+		firstviewtest.Serve(t, ctx, js, ctl, delivery.IndexerModule, st, readers...)
+	}
+	firstviewtest.Serve(t, ctx, js, ctl, delivery.AssignerModule, delivery.PlacementState, readers...)
 	b.builds = mk(delivery.BuilderBuildsBucket, delivery.BuilderBuildsHistory)
 	b.instances = mk(instancesBucket, 1)
 	// The builder's module log, as the runtime raises it (ADR 0297): its stream on the log's subjects.
@@ -791,7 +799,12 @@ func TestTheBuilderAsksTheBuildSeatsAndCallsNoNode(t *testing.T) {
 	}
 	defer sub.Close()
 	var calls atomic.Int64
-	s, err := sub.Subscribe("mesh.mod.*.call.>", func(*nats.Msg) { calls.Add(1) })
+	// A build module's tool, never the owners' read tools the builder's first views ask (DLV-M1).
+	s, err := sub.Subscribe("mesh.mod.*.call.>", func(m *nats.Msg) {
+		if strings.HasPrefix(m.Subject, "mesh.mod.mesh-build-") {
+			calls.Add(1)
+		}
+	})
 	if err != nil {
 		t.Fatal(err)
 	}

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

@@ -25,6 +25,9 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/process"
+	"git.novox.be/novox/mesh-sdk/go/views"
+
+	"git.novox.be/novox/mesh-delivery/sdk/firstview"
 
 	"mesh-delivery-builder/internal/build"
 )
@@ -61,16 +64,23 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	if err != nil {
 		return err
 	}
-	modules, err := process.EssentialBucket(ctx, js, "builder", delivery.ModulesBucket, "know any fingerprint to build")
+	// The indexer's modules, manifests and candidates and the assigner's placement are their owners' states, read by the
+	// first view of their read tools and then their views families, never their buckets (ADR 0615 §1, DLV-M1). An owner that
+	// is down never holds the start: each is read on a goroutine once its first view is answered, and until then the
+	// unanswered first view is the builder's finding.
+	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
 	if err != nil {
 		return err
 	}
-	manifests, err := process.EssentialBucket(ctx, js, "builder", delivery.ManifestsBucket, "read what a module builds")
-	if err != nil {
-		return err
-	}
-	// The candidates' fingerprints are built too (issue 1684): a builder that cannot read them builds the trunk's.
-	candidates := process.OptionalBucket(ctx, js, delivery.CandidatesBucket, "no candidate's fingerprint is built until the builder starts again; the trunk's are", logger.Printf)
+	defer calls.Close()
+	reader := &firstview.Reader{Module: delivery.BuilderModule, Call: calls.Call, Logf: logger.Printf}
+	findings := process.AwaitBucket(ctx, js, conn, findingsBucket, "a first view unanswered past its bound is said in the log alone", logger.Printf)
+	reader.Findings = findings.KV
+	modules := reader.Follow(delivery.IndexerModule, modulesState)
+	manifests := reader.Follow(delivery.IndexerModule, manifestsState)
+	// The candidates' fingerprints are built too (issue 1684).
+	candidates := reader.Follow(delivery.IndexerModule, candidatesState)
+	placement := reader.Follow(delivery.AssignerModule, delivery.PlacementState)
 	bound := defaultStepBound
 	if v := strings.TrimSpace(getenv("MESH_BUILDER_STEP_BOUND")); v != "" {
 		if bound, err = time.ParseDuration(v); err != nil {
@@ -103,12 +113,7 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	heard := buildrun.HearEnds(ctx, asks, asks, buildModules, logger.Printf)
 	conn.SetReconnectHandler(func(*nats.Conn) { hearing.Again(); heard.Nudge("") })
 	// The builder asks the controller, as its own account, whether the artifact store still holds a done key's blobs
-	// (the seat verb held, novox/hq issue 1937) before it leaves the key done.
-	calls, err := process.Connect(cred, nats.Name(cred.Node+"."+cred.Module+".calls"), nats.RetryOnFailedConnect(true))
-	if err != nil {
-		return err
-	}
-	defer calls.Close()
+	// (the seat verb held, novox/hq issue 1937) before it leaves the key done, on its calls connection.
 	b := &build.Builder{
 		Claimer: build.Claimer{Store: kvStore{builds}, By: cred.Node + "." + cred.Module},
 		Asks:    buildrun.Asker{Publisher: asks, Waiter: buildrun.LostAskWaiter{Inner: heard, Dead: dead, Hearing: hearing}, Bound: bound},
@@ -165,10 +170,21 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 	// placement is settled first, so the toolchains settle through the gate already knowing it.
 	gate := buildrun.NewToolchainGate([]string{buildrun.GoBuildModule, buildrun.NodeBuildModule}, tools.Put, logger.Printf)
 	watch.put = gate.Toolchain
-	placementErrs, err := buildrun.FollowPlacement(ctx, js, gate, b.NodesChanged, logger.Printf)
-	if err != nil {
-		return err
-	}
+	// Until the placement's first view is answered the gate passes every report on, as before the placement is settled.
+	placementErrs := make(chan error, 1)
+	go func() {
+		select {
+		case <-ctx.Done():
+			return
+		case <-placement.Follower().Ready():
+		}
+		pe, err := buildrun.FollowPlacement(ctx, pictures{placement}, gate, b.NodesChanged, logger.Printf)
+		if err != nil {
+			placementErrs <- err
+			return
+		}
+		placementErrs <- <-pe
+	}()
 	if err := watch.start(); err != nil {
 		return err
 	}
@@ -202,6 +218,78 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		}
 		defer cc.Stop()
 	}
+	reader.Start(ctx, js, cred, process.AwaitBucket(ctx, js, conn, firstViewsBucket, "builder_first_views reads nothing and builder_first_view_again is not heard", logger.Printf))
+	// The modules, once their first views and the manifests' are answered: what a module builds is read from the manifests.
+	go func() { errs <- buildFromModules(ctx, b, modules, manifests, candidates) }()
+	// A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as
+	// every other input. The builder asks again by itself without it.
+	// A state not on the bus yet at this start is watched once it appears (issue 1790).
+	process.AwaitBucket(ctx, js, conn, build.ReAsksBucket, "builder_ask_again is not heard; an ask that can no longer be answered is still asked again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) {
+		eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) }
+		if rw, err := settle(ctx, kv, eachReAsk); err != nil {
+			process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, rw, eachReAsk) }()
+		}
+	})
+	// A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the
+	// builder alone may read the stream, so it answers there.
+	process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) {
+		eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) }
+		if qw, err := settle(ctx, kv, eachRead); err != nil {
+			process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again")
+		} else {
+			go func() { errs <- follow(ctx, qw, eachRead) }()
+		}
+	})
+	logger.Printf("started on %s", cred.Node)
+	select {
+	case <-ctx.Done():
+		b.Wait()
+		return nil
+	case err := <-errs:
+		return err
+	case err := <-placementErrs:
+		return err
+	case err := <-watch.errs:
+		return err
+	}
+}
+
+// The builder's own states of its first views (views.StatusState) and of its findings, where the condition keeper raises a
+// first view unanswered past its bound (ADR 0615 §6).
+var (
+	firstViewsBucket = delivery.BuilderModule + "_" + views.StatusState
+	findingsBucket   = delivery.BuilderModule + "_findings"
+)
+
+// The indexer's states the builder follows (delivery.ModulesBucket, ManifestsBucket, CandidatesBucket).
+const (
+	modulesState    = "modules"
+	manifestsState  = "manifests"
+	candidatesState = "candidates"
+)
+
+// pictures opens the placement's picture where the gate's follower opens its bucket.
+type pictures struct{ placement *firstview.Picture }
+
+func (p pictures) KeyValue(ctx context.Context, bucket string) (jetstream.KeyValue, error) {
+	if bucket != p.placement.Bucket() {
+		return nil, fmt.Errorf("the builder reads %s by its owner's first view alone", bucket)
+	}
+	return p.placement, nil
+}
+
+// buildFromModules builds from the indexer's modules and candidates once the first views of the modules, the manifests and
+// the candidates are answered (never at the start: an indexer that is down holds no other part of it), and follows them.
+func buildFromModules(ctx context.Context, b *build.Builder, modules, manifests, candidates *firstview.Picture) error {
+	for _, p := range []*firstview.Picture{modules, manifests, candidates} {
+		select {
+		case <-ctx.Done():
+			return nil
+		case <-p.Follower().Ready():
+		}
+	}
 	each := func(e jetstream.KeyValueEntry) {
 		if opOf(e) {
 			b.Module(ctx, e.Key(), e.Value())
@@ -225,48 +313,12 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			b.ForgetCandidate(e.Key())
 		}
 	}
-	if candidates != nil {
-		cw, err := settle(ctx, candidates, eachCandidate)
-		if err != nil {
-			process.NotWatched(logger.Printf, delivery.CandidatesBucket, err, "no candidate's fingerprint is built until the builder starts again")
-		} else {
-			go func() { errs <- follow(ctx, cw, eachCandidate) }()
-		}
-	}
-	// A person asks a key's build again with builder_ask_again, which writes the reasks state (issue 1945): watched, as
-	// every other input. The builder asks again by itself without it.
-	// A state not on the bus yet at this start is watched once it appears (issue 1790).
-	process.AwaitBucket(ctx, js, conn, build.ReAsksBucket, "builder_ask_again is not heard; an ask that can no longer be answered is still asked again by itself", logger.Printf).Ready(func(kv jetstream.KeyValue) {
-		eachReAsk := func(e jetstream.KeyValueEntry) { reAsked(ctx, b, kv, e, logger.Printf) }
-		if rw, err := settle(ctx, kv, eachReAsk); err != nil {
-			process.NotWatched(logger.Printf, build.ReAsksBucket, err, "builder_ask_again is not heard until the builder starts again")
-		} else {
-			go func() { errs <- follow(ctx, rw, eachReAsk) }()
-		}
-	})
-	// A person reads a build seat's stream with builder_seat_queue, which writes the queues state (issue 1945 note 62): the
-	// builder alone may read the stream, so it answers there.
-	process.AwaitBucket(ctx, js, conn, build.QueuesBucket, "builder_seat_queue is not answered", logger.Printf).Ready(func(kv jetstream.KeyValue) {
-		eachRead := func(e jetstream.KeyValueEntry) { queueRead(ctx, heldOnTheBus{js}, kv, e, logger.Printf) }
-		if qw, err := settle(ctx, kv, eachRead); err != nil {
-			process.NotWatched(logger.Printf, build.QueuesBucket, err, "builder_seat_queue is not answered until the builder starts again")
-		} else {
-			go func() { errs <- follow(ctx, qw, eachRead) }()
-		}
-	})
-	logger.Printf("started on %s", cred.Node)
-	go func() { errs <- follow(ctx, w, each) }()
-	select {
-	case <-ctx.Done():
-		b.Wait()
-		return nil
-	case err := <-errs:
-		return err
-	case err := <-placementErrs:
-		return err
-	case err := <-watch.errs:
+	cw, err := settle(ctx, candidates, eachCandidate)
+	if err != nil {
 		return err
 	}
+	go func() { _ = follow(ctx, cw, eachCandidate) }()
+	return follow(ctx, w, each)
 }
 
 // sourceHeads are the trunk heads of each revision of the gate's source.main the watch delivered, the newest

modules/mesh-delivery-builder/go.mod

@@ -25,3 +25,7 @@ require (
 )
 
 replace git.novox.be/novox/mesh-delivery/modules/mesh-delivery-builder/buildrun => ./buildrun
+
+require git.novox.be/novox/mesh-delivery/sdk v0.0.0-00010101000000-000000000000
+
+replace git.novox.be/novox/mesh-delivery/sdk => ../../sdk

modules/mesh-delivery-builder/manifest_test.go

@@ -93,13 +93,13 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	if m.Module != "mesh-delivery-builder" {
 		t.Errorf("module = %q", m.Module)
 	}
-	if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
+	if want := []state{{Name: "builds", History: 5}, {Name: "hearing", History: 1}, {Name: "instances", History: 1}, {Name: "queues", History: 1}, {Name: "reasks", History: 1}, {Name: "first-views", History: 1}, {Name: "findings", History: 1}}; !reflect.DeepEqual(m.State, want) { // delivery.BuilderBuildsHistory
 		t.Errorf("state = %+v, want %+v", m.State, want)
 	}
 	// 1887/A4c: the builder chooses no node, so it reads neither presence nor have, invokes no build module's tool and
 	// holds no capacity; it puts asks on the three build seats and reads their steps (ADR 0525 §1, §4, §5).
 	if want := []string{"mesh-build-go.steps", "mesh-build-go.toolchains", "mesh-build-image.steps", "mesh-build-node.steps", "mesh-build-node.toolchains",
-		"mesh-delivery-assigner.placement", "mesh-delivery-gate.source", "mesh-delivery-indexer.candidates", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules"}; !reflect.DeepEqual(sorted(m.Reads), want) {
+		"mesh-delivery-gate.source"}; !reflect.DeepEqual(sorted(m.Reads), want) {
 		t.Errorf("reads = %v, want %v", m.Reads, want)
 	}
 	if want := []string{"node-build-go", "node-build-image", "node-build-node"}; !reflect.DeepEqual(sorted(m.Uses), want) {
@@ -108,13 +108,22 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 	// 1937/AC3: the builder asks the controller whether the artifact store still holds a done key's blobs (held), and no
 	// other tool; it holds no capacity.
 	// 1955/L7 (ADR 0571 §2): the seat mesh-artifact-sweep's held, the controller's `held` is no longer asked (issue 1955).
-	if want := []string{"seat:mesh-artifact-sweep.held"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
+	// DLV-M1 (ADR 0615 §1): the indexer's modules, manifests and candidates and the assigner's placement are asked by their
+	// read tools and followed by their views families, never read.
+	for _, r := range m.Reads {
+		if strings.HasPrefix(r, "mesh-delivery-indexer.") || strings.HasPrefix(r, "mesh-delivery-assigner.") {
+			t.Errorf("reads %s: the indexer's and the assigner's state is read by their tools and views families alone", r)
+		}
+	}
+	if want := []string{"mesh-delivery-assigner.assigner_read_placement", "mesh-delivery-indexer.indexer_read_candidates", "mesh-delivery-indexer.indexer_read_manifests",
+		"mesh-delivery-indexer.indexer_read_modules", "seat:mesh-artifact-sweep.held"}; !reflect.DeepEqual(sorted(m.Invokes), want) || len(m.Config) != 0 {
 		t.Errorf("invokes = %v, config = %v, want %v and no config", m.Invokes, m.Config, want)
 	}
 	// The builder is woken by a build module's report of a toolchain, as the indexer is: the bucket may not exist when it
 	// starts; and by the seat mesh-artifact-sweep's artifacts-gone, a blob the sweep found gone (1937/AC4, 1955/L7), the
 	// controller's no longer (1126/B9, ADR 0590).
-	if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
+	if want := []string{"mesh-artifact-sweep.artifacts-gone", "mesh-build-go.toolchain.reported", "mesh-build-node.toolchain.reported",
+		"mesh-delivery-assigner.views.placement", "mesh-delivery-indexer.views.candidates", "mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules"}; !reflect.DeepEqual(sorted(m.Consumes), want) {
 		t.Errorf("consumes = %v, want %v", m.Consumes, want)
 	}
 	if b, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 || b != "${dir:state}/broker" {
@@ -124,7 +133,7 @@ func TestManifestDeclarations(t *testing.T) { // 1607/AC1
 		t.Errorf("secrets-owner = %q", m.SecretsOwner)
 	}
 	// 1945: the builder's tools (builder_seat_queue: note 62), served by its tools bundle, and the event it says when it asks an ask again.
-	if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) {
+	if want := []string{"builder_announce_again", "builder_ask_again", "builder_build_again", "builder_builds_since", "builder_first_view_again", "builder_first_views", "builder_seat_queue", "builder_stuck"}; !reflect.DeepEqual(sorted(m.Tools), want) {
 		t.Errorf("tools = %v, want %v", m.Tools, want)
 	}
 	if want := []string{"build.asked-again"}; !reflect.DeepEqual(m.Emits, want) {

modules/mesh-delivery-builder/module.json

@@ -8,7 +8,9 @@
     "builder_build_again",
     "builder_builds_since",
     "builder_seat_queue",
-    "builder_stuck"
+    "builder_stuck",
+    "builder_first_views",
+    "builder_first_view_again"
   ],
   "emits": [
     "build.asked-again"
@@ -16,17 +18,27 @@
   "consumes": [
     "mesh-artifact-sweep.artifacts-gone",
     "mesh-build-go.toolchain.reported",
-    "mesh-build-node.toolchain.reported"
+    "mesh-build-node.toolchain.reported",
+    "mesh-delivery-assigner.views.placement",
+    "mesh-delivery-indexer.views.candidates",
+    "mesh-delivery-indexer.views.manifests",
+    "mesh-delivery-indexer.views.modules"
   ],
   "invokes": [
-    "seat:mesh-artifact-sweep.held"
+    "seat:mesh-artifact-sweep.held",
+    "mesh-delivery-assigner.assigner_read_placement",
+    "mesh-delivery-indexer.indexer_read_candidates",
+    "mesh-delivery-indexer.indexer_read_manifests",
+    "mesh-delivery-indexer.indexer_read_modules"
   ],
   "state": [
     { "name": "builds", "history": 5 },
     { "name": "hearing", "history": 1 },
     { "name": "instances", "history": 1 },
     { "name": "queues", "history": 1 },
-    { "name": "reasks", "history": 1 }
+    { "name": "reasks", "history": 1 },
+    { "name": "first-views", "history": 1 },
+    { "name": "findings", "history": 1 }
   ],
   "logs": [
     { "name": "build-log", "max-age-days": 30 }
@@ -37,11 +49,7 @@
     "mesh-build-image.steps",
     "mesh-build-node.steps",
     "mesh-build-node.toolchains",
-    "mesh-delivery-assigner.placement",
-    "mesh-delivery-gate.source",
-    "mesh-delivery-indexer.candidates",
-    "mesh-delivery-indexer.manifests",
-    "mesh-delivery-indexer.modules"
+    "mesh-delivery-gate.source"
   ],
   "uses": [
     "node-build-go",