DLV-F2r, DLV-F3r (assigner): the assigner reads the indexer's manifests and modules through its first views

The assigner no longer reads the indexer's manifests and modules buckets
(ADR 0615 §1): it asks mesh-delivery-indexer.manifests and .modules at its
start, its first pass waiting for both answers as it waits for every watch,
then follows indexer-views on its durable consumer, asking once more on a
gap; past the bound its finding names the indexer and the tool. Its status
is in its state first-views, read by assigner_first_views and asked again
by hand with assigner_first_view_again. The whole-module tests run against
an indexer owner built from the views emitter and first-view answer.

Issue 1966.

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

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

@@ -102,6 +102,8 @@ func startBed(t *testing.T, leaveOut string) *bed {
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "CONDITIONS", Subjects: []string{"mesh.seat.operator-channel.accept.condition.>"}}); err != nil {
 		t.Fatal(err)
 	}
+	// The indexer as the owner of its manifests and modules (DLV-F2r), with the real owner's side.
+	serveIndexer(t, b, nc)
 	// The assigner's views family and its marks (ADR 0615 §5, ADR 0619 §2), on its old sections.
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "ASSIGNER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-assigner.event.views.>"}}); err != nil {
 		t.Fatal(err)
@@ -1266,3 +1268,49 @@ func placementSubjects() []string {
 	}
 	return out
 }
+
+// serveIndexer is the indexer as the owner of its manifests and modules (ADR 0615, ADR 0621): its buckets, which a test
+// fills as the indexer writes them; its views family said by the views emitter onto INDEXER_VIEWS, read by the
+// assigner's durable consumer; and its first-view tools manifests and modules answered from the buckets at their marks.
+func serveIndexer(t *testing.T, b *bed, nc *nats.Conn) {
+	t.Helper()
+	marks, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: delivery.IndexerModule + "_" + views.MarksState, History: 1})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := b.js.CreateStream(b.ctx, jetstream.StreamConfig{Name: "INDEXER_VIEWS", Subjects: []string{"mesh.stream.mesh-delivery-indexer.event.views.>"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := b.js.CreateOrUpdateConsumer(b.ctx, "INDEXER_VIEWS", jetstream.ConsumerConfig{Durable: "n1_mesh-delivery-assigner", AckPolicy: jetstream.AckExplicitPolicy}); err != nil {
+		t.Fatal(err)
+	}
+	em := views.Emitter{Owner: delivery.IndexerModule, Subject: func(e string) string { return process.ModuleStreamSubject(delivery.IndexerModule, e) },
+		Publish: views.JetStreamPublish(b.js), Marks: marks}
+	for state, bucket := range map[string]string{"manifests": delivery.ManifestsBucket, "modules": delivery.ModulesBucket} {
+		kv, err := b.js.CreateKeyValue(b.ctx, jetstream.KeyValueConfig{Bucket: bucket, History: 5})
+		if err != nil {
+			t.Fatal(err)
+		}
+		b.buckets[bucket] = kv
+		if err := em.Start(b.ctx, state, kv, nil); err != nil {
+			t.Fatal(err)
+		}
+		sub, err := nc.Subscribe("mesh.mod.mesh-delivery-indexer.call."+state+".>", func(m *nats.Msg) {
+			var args struct {
+				After string `json:"after"`
+			}
+			_ = json.Unmarshal(m.Data, &args)
+			a, err := views.FirstView(delivery.IndexerModule, state, bedSource{b.ctx, kv}, bedSource{b.ctx, marks}, args.After, time.Now())
+			out := map[string]any{"result": a, "node": "n0"}
+			if err != nil {
+				out = map[string]any{"error": err.Error()}
+			}
+			raw, _ := json.Marshal(out)
+			_ = m.Respond(raw)
+		})
+		if err != nil {
+			t.Fatal(err)
+		}
+		t.Cleanup(func() { _ = sub.Unsubscribe() })
+	}
+}

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

@@ -58,8 +58,6 @@ var readBuckets = []string{
 	delivery.PartGrantsBucket,
 	delivery.AnswersBucket,
 	delivery.BuilderBuildsBucket,
-	delivery.ManifestsBucket,
-	delivery.ModulesBucket,
 	delivery.PresenceBucket,
 	delivery.HaveBucket,
 	provisions.ProvisionsBucket,
@@ -101,6 +99,7 @@ var keptStates = []announce.State{
 	{Name: delivery.AssignerPoliciesState, Readable: true},
 	{Name: delivery.SeatHoldersState, Readable: true},
 	{Name: views.MarksState, Readable: true}, // what its views family said each state up to (views.go)
+	{Name: views.StatusState, Readable: true}, // its first views of the indexer's states (views.go)
 }
 
 // buckets is every bucket the assigner opened, by name.
@@ -223,7 +222,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 			conn.Close()
 		}
 	}()
-	return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger)
+	iv := &indexerViews{js: js, account: account, call: said.Call, findings: findings, logf: logger.Printf,
+		status: process.AwaitBucket(ctx, js, conn, firstViewsBucket, "assigner_first_views reads nothing and assigner_first_view_again is not heard", logger.Printf)}
+	return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, iv, logger)
 }
 
 // findingsOf is where the assigner's said conditions go as findings (issue 1952 WM6): its findings bucket, which the
@@ -258,7 +259,6 @@ var watched = []string{
 	delivery.PartGrantsBucket,
 	delivery.AnswersBucket,
 	delivery.BuilderBuildsBucket,
-	delivery.ModulesBucket,
 	delivery.PlacementBucket,
 	delivery.PlacementProposalsBucket,
 	delivery.PlacementCandidatesBucket,
@@ -278,7 +278,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, iv *indexerViews, logger *log.Logger) error {
 	w := newWorld()
 	dirty := make(chan struct{}, 1)
 	wake := func() {
@@ -377,8 +377,10 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 			return err
 		}
 	}
-	if err := watch(delivery.ManifestsBucket, jetstream.MetaOnly()); err != nil {
-		return err
+	// The indexer's manifests and modules, through its first views and views family, never its buckets (DLV-F2r): the
+	// first pass waits for both answers, as it waits for every watch to hand what stands.
+	if iv != nil {
+		started += iv.start(ctx, w, wake, settled)
 	}
 	// The seat's states arrive with the same delivery as this assigner (ADR 0567 §1): one not on the bus at this start is
 	// bound once the bus says it was created, and watched from then on, with no restart (issue 1790's kind). Until then no
@@ -410,7 +412,7 @@ func follow(ctx context.Context, js jetstream.JetStream, nc process.Subscriber,
 		logger.Printf("the bucket %s is open and watched", a.name)
 		return watch(a.name)
 	}
-	for range started + 1 {
+	for range started {
 		select {
 		case <-ctx.Done():
 			return nil

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

@@ -57,7 +57,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
 	for _, pair := range []struct {
 		what       string
 		code, want []string
-	}{{"reads", readBuckets, reads}, {"states", append(slices.Clone(ownBuckets), marksBucket), states}} { // the marks are bound by the views family (views.go)
+	}{{"reads", readBuckets, reads}, {"states", append(slices.Clone(ownBuckets), marksBucket, firstViewsBucket), states}} { // the marks are bound by the views family (views.go)
 		got, want := slices.Sorted(slices.Values(pair.code)), slices.Sorted(slices.Values(pair.want))
 		if !reflect.DeepEqual(got, want) {
 			t.Errorf("the program's %s are %v, the manifest's %v", pair.what, got, want)
@@ -67,7 +67,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
 	for _, s := range keptStates {
 		kept = append(kept, m.Module+"_"+s.Name)
 	}
-	if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(append(slices.Clone(ownBuckets), marksBucket))); !reflect.DeepEqual(got, want) {
+	if got, want := slices.Sorted(slices.Values(kept)), slices.Sorted(slices.Values(append(slices.Clone(ownBuckets), marksBucket, firstViewsBucket))); !reflect.DeepEqual(got, want) {
 		t.Errorf("the program announces %v, keeps %v", got, want)
 	}
 	// ADR 0546: whatever keeps the puts' bucket from opening (not there, not granted, no answer in time), the assigner opens
@@ -113,7 +113,7 @@ func TestOpensWhatItsManifestNames(t *testing.T) {
 	}
 	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/pass.go

@@ -107,7 +107,7 @@ func (p *passer) nodeWants(ctx context.Context) {
 // the hold of a fingerprint whose health check failed. The assigner compares its wants with nothing the controller
 // sends: the differences bucket is kept read-only with its history (ADR 0539 §8).
 func (p *passer) wants(ctx context.Context) {
-	get := kvGetter(p.kvs[delivery.ManifestsBucket])
+	get := p.w.getter(delivery.ManifestsBucket) // the indexer's manifests, as its first view and views family say them
 	assigned := p.w.assigned()
 	trunk := p.w.trunk()
 	manifests := map[string]catalogue.Manifest{}

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

@@ -99,7 +99,7 @@ func (p *passer) puts(ctx context.Context) {
 	}
 	assigned := p.w.assigned()
 	trunk := p.w.trunk()
-	get := kvGetter(p.kvs[delivery.ManifestsBucket])
+	get := p.w.getter(delivery.ManifestsBucket) // the indexer's manifests, as its first view and views family say them
 	wanted := p.w.wantsOf()
 	modules := make([]string, 0, len(all))
 	for m := range all {

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

@@ -2,11 +2,13 @@ package main
 
 import (
 	"context"
+	"encoding/json"
 	"time"
 
 	"github.com/nats-io/nats.go/jetstream"
 
 	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
 	"git.novox.be/novox/mesh-sdk/go/health"
 	"git.novox.be/novox/mesh-sdk/go/process"
@@ -64,7 +66,8 @@ func viewTools(getenv func(string) string, state func(string) views.Source, now
 	for _, t := range readTools {
 		out = append(out, views.ReadTool(delivery.AssignerModule, t.tool, t.state, t.words, state(t.state), marks, now))
 	}
-	return out
+	// Its own first views, of the indexer's manifests and modules (DLV-F3r): read, and asked again by hand.
+	return append(out, views.StatusTools(delivery.AssignerModule, "assigner_first_views", "assigner_first_view_again", stdio.State(views.StatusState), now)...)
 }
 
 func keptSource(name string) views.Source { return views.KeptSource{State: stdio.State(name)} }
@@ -101,3 +104,77 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib
 		}
 	})
 }
+
+// firstViewsBucket is the assigner's own state of its first views (views.StatusState).
+var firstViewsBucket = delivery.AssignerModule + "_" + views.StatusState
+
+// indexerViews are the assigner's first views of the indexer's manifests and modules (ADR 0615 §1, ADR 0621 §2;
+// DLV-F2r): asked at the start through the indexer's tools, then followed on indexer-views, asking once more on a gap;
+// past the bound its finding names the indexer and the tool, cleared by the answer.
+type indexerViews struct {
+	js       jetstream.JetStream
+	account  process.Account
+	call     func(string, any) (json.RawMessage, error)
+	findings assign.Findings
+	status   *process.AwaitedBucket
+	logf     func(string, ...any)
+}
+
+// start asks both first views, each answer taken into w under the indexer's bucket names and settled once (a pass waits
+// for it), and hears the views family; it answers how many settles to wait for.
+func (iv *indexerViews) start(ctx context.Context, w *world, wake func(), settled chan<- struct{}) int {
+	finding := func(s views.Status, raised bool) {
+		if iv.findings.Put == nil {
+			return
+		}
+		key, value, err := views.UnansweredFinding(delivery.AssignerModule, s, time.Now())
+		if err == nil && raised {
+			err = iv.findings.Put(key, value)
+		} else if err == nil {
+			err = iv.findings.Delete(key)
+		}
+		if err != nil {
+			iv.logf("views: the finding of the first view %s is not written: %v", s.Tool, err)
+		}
+	}
+	followers := map[string]*views.Follower{}
+	var all []*views.Follower
+	for state, bucket := range map[string]string{"manifests": delivery.ManifestsBucket, "modules": delivery.ModulesBucket} {
+		bucket := bucket
+		tool := delivery.IndexerModule + "." + state
+		f := &views.Follower{Owner: delivery.IndexerModule, Tool: tool, State: state, Ask: views.AskVia(iv.call, tool), Logf: iv.logf,
+			Raise: func(s views.Status) { finding(s, true) }, Clear: func(s views.Status) { finding(s, false) },
+			Changed: func(key string, value json.RawMessage, deleted bool, at time.Time) {
+				if deleted {
+					value = nil
+				}
+				w.putEntry(bucket, key, value, 0, at)
+				wake()
+			}}
+		followers[state] = f
+		all = append(all, f)
+		go func() {
+			if f.Start(ctx) == nil {
+				settled <- struct{}{}
+			}
+		}()
+	}
+	if iv.status != nil {
+		iv.status.Ready(func(kv jetstream.KeyValue) {
+			for _, f := range all {
+				f.Said = views.KeepStatus(ctx, kv, iv.logf)
+			}
+			if err := views.HearAgain(ctx, kv, all, iv.logf); err != nil {
+				iv.logf("views: assigner_first_view_again is not heard (%v) until the assigner starts again", err)
+			}
+		})
+	}
+	stream := catalogue.StreamBusName(views.IndexerStream)
+	if cons, err := iv.account.EventConsumer(ctx, iv.js, stream); err != nil {
+		iv.logf("the consumer %s on %s does not exist yet (%v): the indexer's manifests and modules are the first views' alone until the assigner starts again",
+			iv.account.EventConsumerName(), stream, err)
+	} else if _, err := process.HearEvents(cons, views.Hear(ctx, followers, iv.logf), iv.logf); err != nil {
+		iv.logf("the indexer's views family is not heard (%v) until the assigner starts again", err)
+	}
+	return len(all)
+}

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

@@ -197,6 +197,19 @@ func kvGetter(kv jetstream.KeyValue) getter {
 	}
 }
 
+// getter reads a key of a bucket the world holds: errNotThere where it holds none.
+func (w *world) getter(bucket string) getter {
+	return func(_ context.Context, key string) ([]byte, error) {
+		w.mu.Lock()
+		defer w.mu.Unlock()
+		v, ok := w.values[bucket][key]
+		if !ok {
+			return nil, errNotThere
+		}
+		return v, nil
+	}
+}
+
 // manifest reads the manifest of a digest from the indexer's bucket once and keeps it: a digest's manifest never changes.
 func (w *world) manifest(ctx context.Context, get getter, digest string) parsed {
 	w.mu.Lock()

modules/mesh-delivery-assigner/manifest_test.go

@@ -38,7 +38,8 @@ func TestManifest(t *testing.T) {
 	// ADR 0593 §8: it composes nothing, so it reads no candidate, held or resolved value, compose step or build log; it reads
 	// every node's published compositions to compare them, and the gate's source and merges, for the release that carries a build (ADR 0601 §1);
 	// and the indexer's module graph, whose verb edges a want waits on (ADR 0614 §2, §3).
-	want := []string{"mesh-delivery-builder.builds", "mesh-delivery-gate.merges", "mesh-delivery-gate.source", "mesh-delivery-indexer.graph", "mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
+	// DLV-F2r (ADR 0615 §1): the indexer's manifests and modules are read through its first views and views family.
+	want := []string{"mesh-delivery-builder.builds", "mesh-delivery-gate.merges", "mesh-delivery-gate.source", "mesh-delivery-indexer.graph",
 		"mesh-delivery-teller.answers", "mesh-delivery-watchman.presence", "mesh-grants.grants", "mesh-host.composed",
 		"mesh-provisioner.provisions", "node-engine.have"}
 	if !reflect.DeepEqual(reads, want) {
@@ -49,7 +50,7 @@ func TestManifest(t *testing.T) {
 		states[s.Name] = s.History
 	}
 	if want := map[string]int{"asks": 5, "differences": 64, "puts": 20, "wants": 20,
-		"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1}; !reflect.DeepEqual(states, want) {
+		"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1, views.StatusState: 1}; !reflect.DeepEqual(states, want) {
 		t.Errorf("state = %v, want %v", states, want)
 	}
 	if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
@@ -83,7 +84,8 @@ func TestManifest(t *testing.T) {
 	// ADR 0621 §1: the first-view tools placement, seat-holders and wants and design 25 § 12.2's six read tools, of the
 	// program's bundle, which runs in a runtime of its own on the assigner's account.
 	if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts", "placement", "seat-holders", "wants",
-		"assigner_read_placement", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences"}) {
+		"assigner_read_placement", "assigner_read_seat_holders", "assigner_read_wants", "assigner_read_findings", "assigner_read_asks", "assigner_read_differences",
+		"assigner_first_views", "assigner_first_view_again"}) {
 		t.Errorf("the assigner serves tools %v, want its three put tools, three first views and six read tools", m.Tools)
 	}
 	if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs {
@@ -106,7 +108,10 @@ func TestManifest(t *testing.T) {
 		!slices.Equal(slices.Sorted(slices.Values(m.Streams[0].Events)), slices.Sorted(slices.Values(delivery.PlacementEvents))) {
 		t.Errorf("streams = %+v, want placement carrying the seat's four events for %d days", m.Streams, delivery.PlacementStreamDays)
 	}
-	if want := []string{
+	if !slices.Equal(m.Consumes, []string{"mesh-delivery-indexer.views.manifests", "mesh-delivery-indexer.views.modules"}) {
+		t.Errorf("consumes = %v, want the indexer's views of manifests and modules", m.Consumes)
+	}
+	if want := []string{"mesh-delivery-indexer.manifests", "mesh-delivery-indexer.modules",
 		"seat:mesh-hand-acts.hand-act", "seat:mesh-hand-acts.hand-acts", "seat:mesh-hand-acts.warranted", // ADR 0565 §9, §11: the seat alone
 		"seat:node-backup.backed-up", "seat:node-backup.now"}; !slices.Equal(slices.Sorted(slices.Values(m.Invokes)), want) {
 		t.Errorf("invokes = %v, want %v", m.Invokes, want)

modules/mesh-delivery-assigner/module.json

@@ -29,8 +29,6 @@
     "mesh-delivery-gate.merges",
     "mesh-delivery-gate.source",
     "mesh-delivery-indexer.graph",
-    "mesh-delivery-indexer.manifests",
-    "mesh-delivery-indexer.modules",
     "mesh-delivery-teller.answers",
     "mesh-delivery-watchman.presence",
     "mesh-grants.grants",
@@ -41,7 +39,13 @@
   "uses": [
     "operator-channel"
   ],
+  "consumes": [
+    "mesh-delivery-indexer.views.manifests",
+    "mesh-delivery-indexer.views.modules"
+  ],
   "invokes": [
+    "mesh-delivery-indexer.manifests",
+    "mesh-delivery-indexer.modules",
     "seat:mesh-hand-acts.hand-act",
     "seat:mesh-hand-acts.hand-acts",
     "seat:mesh-hand-acts.warranted",
@@ -63,7 +67,9 @@
     "assigner_read_wants",
     "assigner_read_findings",
     "assigner_read_asks",
-    "assigner_read_differences"
+    "assigner_read_differences",
+    "assigner_first_views",
+    "assigner_first_view_again"
   ],
   "state": [
     { "name": "placement", "history": 64 },
@@ -77,7 +83,8 @@
     { "name": "findings", "history": 5 },
     { "name": "policies", "history": 20 },
     { "name": "seat-holders", "history": 5 },
-    { "name": "view-marks", "history": 1 }
+    { "name": "view-marks", "history": 1 },
+    { "name": "first-views", "history": 1 }
   ],
   "streams": [
     {