DLV-O2, DLV-O3, DLV-O5: the assigner answers its first views and says its views family

The assigner's bundle runs in a runtime of its own on its own account and
serves the first-view tools placement, seat-holders and wants, each answering
only the callers its manifest writes (ADR 0621 §1-§2), and the six read tools
of design 25 § 12.2. The process says each change of those six states on the
family assigner-views beside placement in its old streams section (ADR 0619
§2, ADR 0615 §5); a waiter on put never wakes on it.

Issue 1966.

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

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

@@ -5,6 +5,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"maps"
@@ -30,6 +31,8 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/provisions"
 	"mesh-delivery-assigner/internal/judge"
 
+	"git.novox.be/novox/mesh-delivery/sdk/views"
+
 	"mesh-delivery-assigner/internal/assign"
 )
 
@@ -99,6 +102,15 @@ 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 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)
+	}
+	if kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: marksBucket, History: 1}); err != nil {
+		t.Fatal(err)
+	} else {
+		b.buckets[marksBucket] = kv
+	}
 	// The Go build seat's work queue, as the broker derives it from the seat's accepted kinds (delivery.AskStream).
 	seat, _ := delivery.AskStream(delivery.AskGo)
 	if _, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: seat, Subjects: []string{"mesh.seat.node-build-go.accept.>"},
@@ -376,6 +388,9 @@ func TestWantsAreWrittenAndNoDifferenceIs(t *testing.T) {
 	if w := b.wantNow(t, "shanks", "web", "shanks' first want at what it runs", func(w delivery.Want) bool { return w.FP != "" }); w.FP != fOld || w.Composed != "" || !w.Self || w.Newest != fNew {
 		t.Errorf("shanks' want = %+v, want the fingerprint it runs, the node composing it, held back from F", w)
 	}
+	// DLV-O5 (ADR 0615 §5, ADR 0619 §2): each want is said on the assigner's views family, its value whole at its revision,
+	// and the first view at the mark followed by those events holds it as the bucket does; a waiter on put never wakes.
+	b.viewsHold(t, "wants", wkey)
 	var neither asks.Condition
 	for deadline := time.Now().Add(10 * time.Second); time.Now().Before(deadline) && neither.Key == ""; time.Sleep(100 * time.Millisecond) {
 		for k, c := range b.conditionsSaid(t) {
@@ -1167,3 +1182,78 @@ func TestAProposalIsAnsweredApprovedAndWritten(t *testing.T) {
 		t.Error("the removed node's key stays")
 	}
 }
+
+// viewsHold waits until a follower of state, from a first view at its mark and the views family's events, holds key as
+// the bucket does, and checks that no views event landed where a waiter on the assigner's put waits.
+func (b *bed) viewsHold(t *testing.T, state, key string) {
+	t.Helper()
+	kv, marks := b.buckets[viewedStates[state]], b.buckets[marksBucket]
+	src := func(kv jetstream.KeyValue) views.Source { return bedSource{b.ctx, kv} }
+	st, err := b.js.Stream(b.ctx, "ASSIGNER_VIEWS")
+	if err != nil {
+		t.Fatal(err)
+	}
+	for deadline := time.Now().Add(20 * time.Second); ; time.Sleep(50 * time.Millisecond) {
+		f := &views.Follower{Owner: delivery.AssignerModule, Tool: state, State: state, Ask: func(context.Context, string) (views.Answer, error) {
+			return views.FirstView(delivery.AssignerModule, state, src(kv), src(marks), "", time.Now())
+		}}
+		if err := f.Start(b.ctx); err != nil {
+			t.Fatal(err)
+		}
+		info, _ := st.Info(b.ctx)
+		gap := false
+		for seq := info.State.FirstSeq; seq > 0 && seq <= info.State.LastSeq; seq++ {
+			m, err := st.GetMsg(b.ctx, seq)
+			if err != nil || views.StateOfEvent(m.Subject) != state {
+				continue
+			}
+			e, err := views.DecodeEvent(m.Data)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if g, _ := f.Apply(e); g {
+				gap = true
+			}
+		}
+		e, err := kv.Get(b.ctx, key)
+		v, ok := f.Get(key)
+		if err == nil && ok && !gap && string(v) == string(e.Value()) && f.Status().Revision >= e.Revision() {
+			break
+		}
+		if time.Now().After(deadline) {
+			t.Fatalf("the views family holds %s.%s as %s at %d (gap %v), the bucket %v", state, key, v, f.Status().Revision, gap, err)
+		}
+	}
+	ev, err := b.js.Stream(b.ctx, "EVENTS")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := ev.GetLastMsgForSubject(b.ctx, "mesh.mod.mesh-delivery-assigner.event.put"); err == nil {
+		t.Errorf("a waiter on put woke: the bus holds a put")
+	}
+}
+
+// bedSource is a bucket as a views.Source.
+type bedSource struct {
+	ctx context.Context
+	kv  jetstream.KeyValue
+}
+
+func (s bedSource) Keys() ([]string, error) {
+	k, err := s.kv.Keys(s.ctx)
+	if errors.Is(err, jetstream.ErrNoKeysFound) {
+		return nil, nil
+	}
+	return k, err
+}
+
+func (s bedSource) Get(key string) (*views.Entry, bool, error) {
+	e, err := s.kv.Get(s.ctx, key)
+	if errors.Is(err, jetstream.ErrKeyNotFound) {
+		return nil, false, nil
+	}
+	if err != nil {
+		return nil, false, err
+	}
+	return &views.Entry{Value: e.Value(), Revision: e.Revision()}, true, nil
+}

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

@@ -209,6 +209,18 @@ 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)
+	// Each change of the states its readers read is said on its views family (ADR 0615 §5, ADR 0619 §2); a watch of it
+	// that ends stops the program, as a watch of what a pass reads does.
+	viewErrs := make(chan error, len(viewedStates))
+	startViews(ctx, js, conn, account, kvs, findings, viewErrs, logger.Printf)
+	go func() {
+		select {
+		case <-ctx.Done():
+		case err := <-viewErrs:
+			logger.Printf("%v: the assigner stops, and is started again", err)
+			conn.Close()
+		}
+	}()
 	return follow(ctx, js, conn, kvs, findings.Say(sayVia(ctx, js)), findings, emitting{said}, logger)
 }
 

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", ownBuckets, states}} {
+	}{{"reads", readBuckets, reads}, {"states", append(slices.Clone(ownBuckets), marksBucket), 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)

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

@@ -127,7 +127,8 @@ func TestTheSeatsVerbs(t *testing.T) {
 		}
 	}
 	for _, stream := range raw0.Streams {
-		if !reads["stream "+stream.Name] {
+		// The views family is no seat's stream: the readers of the assigner's states follow it (ADR 0615 §5).
+		if stream.Name != viewsStream && !reads["stream "+stream.Name] {
 			t.Errorf("no verb reads the seat's stream %s", stream.Name)
 		}
 	}

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

@@ -358,6 +358,7 @@ func serveTools() error {
 	})
 	wants.watched = stdio.NewWatchedStates() // the large states read whole from one watched copy (ADR 0580 clause 3)
 	tools = append(tools, wants.tools()...)
+	tools = append(tools, viewTools(os.Getenv, keptSource, time.Now)...)
 	return stdio.Serve("", tools)
 }
 

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

@@ -0,0 +1,103 @@
+package main
+
+import (
+	"context"
+	"time"
+
+	"github.com/nats-io/nats.go/jetstream"
+
+	stdio "git.novox.be/novox/mesh-sdk/go"
+	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/health"
+	"git.novox.be/novox/mesh-sdk/go/process"
+
+	"git.novox.be/novox/mesh-delivery/sdk/views"
+
+	"mesh-delivery-assigner/internal/assign"
+)
+
+// viewsStream is the assigner's views family on its old streams section, beside placement (ADR 0619 §2, ADR 0615 §5).
+const viewsStream = views.AssignerStream
+
+// marksBucket is the assigner's state of what its views family said each state up to.
+var marksBucket = delivery.AssignerModule + "_" + views.MarksState
+
+// viewedTool is one tool over one state of the assigner's.
+type viewedTool struct{ tool, state, words string }
+
+// firstViewTools are the assigner's first-view tools (ADR 0621 §1): each answers only its callers list
+// (views.CallersEnv), which the manifest writes.
+var firstViewTools = []viewedTool{
+	{"placement", delivery.PlacementState, "where each module is placed: the seat mesh-placement's placement, by module."},
+	{"seat-holders", delivery.SeatHoldersState, "who holds which seat on which node."},
+	{"wants", "wants", "what each node should run: want.<node>.<module>, its fingerprint and the rule that gave it."},
+}
+
+// readTools are the assigner's ordinary read tools, one per state its readers read (design 25 § 12.2).
+var readTools = []viewedTool{
+	{"assigner_read_placement", delivery.PlacementState, "where each module is placed, by module."},
+	{"assigner_read_seat_holders", delivery.SeatHoldersState, "who holds which seat on which node."},
+	{"assigner_read_wants", "wants", "what each node should run: want.<node>.<module>."},
+	{"assigner_read_findings", health.FindingsState, "the assigner's findings, which the condition keeper raises and clears."},
+	{"assigner_read_asks", "asks", "the asks the assigner put to a person, and their answers."},
+	{"assigner_read_differences", "differences", "every difference between a want and what a node runs."},
+}
+
+// viewedStates are the states the views family says, each by its bucket: the six its readers read.
+var viewedStates = map[string]string{
+	delivery.PlacementState:   delivery.PlacementBucket,
+	delivery.SeatHoldersState: delivery.SeatHoldersBucket,
+	"wants":                   delivery.AssignerWantsBucket,
+	health.FindingsState:      assign.FindingsBucket,
+	"asks":                    delivery.AssignerAsksBucket,
+	"differences":             delivery.AssignerDifferencesBucket,
+}
+
+// viewTools are the assigner's first-view and read tools, reading only its own states, so they answer as soon as its
+// runtime of its own is connected (ADR 0621 §3).
+func viewTools(getenv func(string) string, state func(string) views.Source, now func() time.Time) []stdio.Tool {
+	marks := state(views.MarksState)
+	var out []stdio.Tool
+	for _, t := range firstViewTools {
+		out = append(out, views.FirstViewTool(delivery.AssignerModule, t.tool, t.state, t.words, views.ParseCallers(getenv(views.CallersEnv(t.tool))), state(t.state), marks, 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
+}
+
+func keptSource(name string) views.Source { return views.KeptSource{State: stdio.State(name)} }
+
+// startViews says each change of the assigner's viewed states on its views family, once its marks state is bound,
+// each state as soon as its bucket is (kvs, or bound once it appears). A watch that ends is sent on errs.
+func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscriber, account process.Account, kvs buckets, findings assign.Findings, errs chan<- error, logf func(string, ...any)) {
+	tooLarge := func(state, key string, size int) {
+		if findings.Put == nil {
+			return
+		}
+		k, v, err := views.TooLargeFinding(delivery.AssignerModule, state, key, size, time.Now())
+		if err == nil {
+			err = findings.Put(k, v)
+		}
+		if err != nil {
+			logf("views: the finding of %s.%s too large is not written: %v", state, key, err)
+		}
+	}
+	process.AwaitBucket(ctx, js, nc, marksBucket, "no change is said on the views family, so every reader waits on its first view", logf).Ready(func(mk jetstream.KeyValue) {
+		em := views.Emitter{Owner: delivery.AssignerModule, Subject: account.StreamSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf}
+		for _, state := range []string{"asks", "differences", health.FindingsState, delivery.PlacementState, delivery.SeatHoldersState, "wants"} {
+			state := state
+			start := func(kv jetstream.KeyValue) {
+				if err := em.Start(ctx, state, kv, errs); err != nil {
+					logf("%v: its readers ask its first view at each change until the assigner starts again", err)
+				}
+			}
+			if kv := kvs[viewedStates[state]]; kv != nil {
+				start(kv)
+				continue
+			}
+			process.AwaitBucket(ctx, js, nc, viewedStates[state], "its changes are not said on the views family", logf).Ready(start)
+		}
+	})
+}

modules/mesh-delivery-assigner/manifest_test.go

@@ -8,6 +8,8 @@ import (
 
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+
+	"git.novox.be/novox/mesh-delivery/sdk/views"
 )
 
 func load(t *testing.T) catalogue.Manifest {
@@ -47,7 +49,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}; !reflect.DeepEqual(states, want) {
+		"placement": 64, "proposals": 5, "candidates": 5, "pulses": 0, "findings": 5, "policies": 20, "seat-holders": 5, views.MarksState: 1}; !reflect.DeepEqual(states, want) {
 		t.Errorf("state = %v, want %v", states, want)
 	}
 	if _, ok := m.OwnSecrets["broker"]; !ok || len(m.OwnSecrets) != 1 {
@@ -78,10 +80,29 @@ func TestManifest(t *testing.T) {
 	if seat, _ := catalogue.SeatNamed(catalogue.PlacementSeat); catalogue.CanHold(m, seat) != nil {
 		t.Errorf("the assigner cannot hold the seat: %v", catalogue.CanHold(m, seat))
 	}
-	if !slices.Equal(m.Tools, []string{"assigner_put", "assigner_release", "assigner_puts"}) {
-		t.Errorf("the assigner serves tools %v, want its three put tools", m.Tools)
+	// 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"}) {
+		t.Errorf("the assigner serves tools %v, want its three put tools, three first views and six read tools", m.Tools)
 	}
-	if len(m.Streams) != 1 || m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays ||
+	if m.RunsAs != m.Module || m.SecretsOwner != m.RunsAs {
+		t.Errorf("runs-as = %q: the bundle runs on the assigner's own account (ADR 0621 §1)", m.RunsAs)
+	}
+	// ADR 0621 §2: each first-view tool answers exactly its readers, never the assigner.
+	callers := map[string]string{"placement": "mesh-grants,node-tools,mesh-delivery-indexer", "seat-holders": "node-tools", "wants": "mesh-grants"}
+	env := m.Build.Artifacts[0].Env
+	for tool, list := range callers {
+		if env[views.CallersEnv(tool)] != list || views.ParseCallers(env[views.CallersEnv(tool)]).Allows("n1."+m.Module) {
+			t.Errorf("%s callers = %q, want %q", tool, env[views.CallersEnv(tool)], list)
+		}
+	}
+	// ADR 0619 §2, ADR 0615 §5: its views family beside placement on its old streams section.
+	if len(m.Streams) != 2 || m.Streams[1].Name != views.AssignerStream || !slices.Equal(m.Streams[1].Events, []string{views.EventPattern}) ||
+		m.Streams[1].MaxAgeDays != views.StreamDays || m.Streams[1].MaxMiB != views.StreamMiB || m.Streams[1].Retention != "" {
+		t.Errorf("streams = %+v, want placement and %s", m.Streams, views.AssignerStream)
+	}
+	if m.Streams[0].Name != delivery.PlacementStream || m.Streams[0].MaxAgeDays != delivery.PlacementStreamDays ||
 		!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)
 	}

modules/mesh-delivery-assigner/module.json

@@ -2,6 +2,7 @@
   "module": "mesh-delivery-assigner",
   "version": "1",
   "slug": "assigner",
+  "runs-as": "mesh-delivery-assigner",
   "claims": [
     {
       "name": "mesh-placement",
@@ -53,7 +54,16 @@
   "tools": [
     "assigner_put",
     "assigner_release",
-    "assigner_puts"
+    "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"
   ],
   "state": [
     { "name": "placement", "history": 64 },
@@ -66,7 +76,8 @@
     { "name": "wants", "history": 20 },
     { "name": "findings", "history": 5 },
     { "name": "policies", "history": 20 },
-    { "name": "seat-holders", "history": 5 }
+    { "name": "seat-holders", "history": 5 },
+    { "name": "view-marks", "history": 1 }
   ],
   "streams": [
     {
@@ -78,7 +89,8 @@
         "placement-withdrawn"
       ],
       "max-age-days": 30
-    }
+    },
+    { "name": "assigner-views", "events": ["views.**"], "max-age-days": 7, "max-mib": 256 }
   ],
   "own-secrets": {
     "broker": "${dir:state}/broker"
@@ -128,7 +140,12 @@
         "binary": "mesh-delivery-assigner",
         "loads": [
           "mesh-delivery-assigner"
-        ]
+        ],
+        "env": {
+          "MESH_VIEW_CALLERS_PLACEMENT": "mesh-grants,node-tools,mesh-delivery-indexer",
+          "MESH_VIEW_CALLERS_SEAT_HOLDERS": "node-tools",
+          "MESH_VIEW_CALLERS_WANTS": "mesh-grants"
+        }
       }
     ]
   }