G-delivery 7: the indexer's and the assigner's views emitters write view-unsaid

mesh-sdk's views.Emitter writes the owner's finding of a change not said past its bound only when Findings is set; both owners of views in mesh-delivery now set it to their findings state. Claude-Session: https://claude.ai/code/session_01JG73KJNrH5EgFXMKKYSrig
2026-11-04 18:20:53 +01:00
parent 24f9b6f602
commit 4ba4cacc66
4 changed files with 47 additions and 3 deletions
@@ -21,11 +21,17 @@ import (
 	oc "git.novox.be/novox/mesh-contracts/go/seats/operatorchannel"
 	"git.novox.be/novox/mesh-sdk/go/catalogue"
 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"
 )

 // 1297/B2a: the buckets the program opens are exactly the manifest's reads and states, as the broker names them
 // (`<module>.<state>` read as the bucket `<module>_<state>`), so it is granted what it reads and reads nothing else.
 func TestOpensWhatItsManifestNames(t *testing.T) {
+	// Review round 2: the assigner's views emitter writes its finding view-unsaid (Emitter.Findings set) when it keeps findings.
+	findings := assign.Findings{Put: func(string, []byte) error { return nil }, Delete: func(string) error { return nil }}
+	if viewsEmitter(nil, process.Account{}, nil, findings, nil, t.Logf).Findings == nil {
+		t.Error("the assigner's views emitter writes no view-unsaid finding")
+	}
 	// 1297/B2a: without a credential the program does not start, and says why.
 	if err := run(context.Background(), func(string) string { return "" }, io.Discard); err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
 		t.Errorf("run without a credential = %v", err)
@@ -98,7 +98,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib
 		}
 	}
 	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}
+		em := viewsEmitter(js, account, mk, findings, tooLarge, logf)
 		for _, state := range []string{"asks", "differences", health.FindingsState, delivery.PlacementState, delivery.SeatHoldersState, "wants"} {
 			state := state
 			start := func(kv jetstream.KeyValue) {
@@ -115,6 +115,22 @@ func startViews(ctx context.Context, js jetstream.JetStream, nc process.Subscrib
 	})
 }

+// viewsEmitter is the assigner's emitter of its views family: a change not said past views.Bound is its finding
+// view-unsaid, in its findings (review round 2); with no findings state it is said in the log only.
+func viewsEmitter(js jetstream.JetStream, account process.Account, mk jetstream.KeyValue, findings assign.Findings, tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter {
+	em := views.Emitter{Owner: delivery.AssignerModule, Subject: account.StreamSubject, Publish: views.JetStreamPublish(js), Marks: mk, TooLarge: tooLarge, Logf: logf}
+	if findings.Put != nil && findings.Delete != nil {
+		em.Findings = assignerFindings{findings}
+	}
+	return em
+}
+
+// assignerFindings is the assigner's findings as views.Emitter's Findings.
+type assignerFindings struct{ f assign.Findings }
+
+func (a assignerFindings) Put(key string, value []byte) error { return a.f.Put(key, value) }
+func (a assignerFindings) Delete(key string) error            { return a.f.Delete(key) }
+
 // firstViewsBucket is the assigner's own state of its first views (views.StatusState).
 var firstViewsBucket = delivery.AssignerModule + "_" + views.StatusState

@@ -12,6 +12,7 @@ import (
 	"github.com/nats-io/nats.go/jetstream"

 	"git.novox.be/novox/mesh-sdk/go/delivery"
+	"git.novox.be/novox/mesh-sdk/go/process"

 	"mesh-delivery-indexer/internal/index"
 )
@@ -71,6 +72,10 @@ func (m *memStore) Delete(_ context.Context, k string, _ uint64) error { delete(
 const head = "0123456789abcdef0123456789abcdef01234567"

 func TestAMessageIsIndexedAndAcked(t *testing.T) { // 1238/AC2 (unit part)
+	// Review round 2: the indexer's views emitter writes its finding view-unsaid (Emitter.Findings set).
+	if em := viewsEmitter(context.Background(), nil, process.Account{}, nil, &process.AwaitedBucket{}, nil, t.Logf); em.Findings == nil {
+		t.Error("the indexer's views emitter writes no view-unsaid finding")
+	}
 	t.Run("not a forge message: terminated and logged", func(t *testing.T) { // 1238 (never redelivered)
 		var logs bytes.Buffer
 		st := &memStore{vals: map[string][]byte{}}
@@ -47,8 +47,7 @@ func startViews(ctx context.Context, js jetstream.JetStream, cred process.Accoun
 		}
 	}
 	marks.Ready(func(mk jetstream.KeyValue) {
-		em := views.Emitter{Owner: delivery.IndexerModule, Subject: cred.StreamSubject, Publish: views.JetStreamPublish(js),
-			Marks: mk, TooLarge: tooLarge, Logf: logf}
+		em := viewsEmitter(ctx, js, cred, mk, findings, tooLarge, logf)
 		for _, state := range sortedStates(states) {
 			states[state](func(kv jetstream.KeyValue) {
 				if err := em.Start(ctx, state, kv, errs); err != nil {
@@ -59,6 +58,24 @@ func startViews(ctx context.Context, js jetstream.JetStream, cred process.Accoun
 	})
 }

+// viewsEmitter is the indexer's emitter of its views family: a change not said past views.Bound is its finding
+// view-unsaid, in its own findings once bound (review round 2).
+func viewsEmitter(ctx context.Context, js jetstream.JetStream, cred process.Account, mk jetstream.KeyValue, findings *process.AwaitedBucket,
+	tooLarge func(string, string, int), logf func(string, ...any)) views.Emitter {
+	w := notHeardWriter(ctx, findings, logf)
+	return views.Emitter{Owner: delivery.IndexerModule, Subject: cred.StreamSubject, Publish: views.JetStreamPublish(js),
+		Marks: mk, TooLarge: tooLarge, Findings: funcFindings{put: w.Put, del: w.Delete}, Logf: logf}
+}
+
+// funcFindings is a findings state as two functions: views.Emitter's Findings.
+type funcFindings struct {
+	put func(string, []byte) error
+	del func(string) error
+}
+
+func (f funcFindings) Put(key string, value []byte) error { return f.put(key, value) }
+func (f funcFindings) Delete(key string) error            { return f.del(key) }
+
 func sortedStates(states map[string]readyOf) []string {
 	out := make([]string, 0, len(states))
 	for s := range states {