The view says its event delivered once for each source revision whose wants and haves are all there (issue 1906 D9, for issue 1946)

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

@@ -696,3 +696,98 @@ func TestProcessWholeShowsTheBacklogOfBuildAsks(t *testing.T) { // 1887/A8a AC
 	}
 	t.Fatalf("the backlog never showed three waiting asks with the oldest's age; last: %+v", last.Backlog)
 }
+
+// 1906/D9, whole on the bus: when the wants and the haves make the source revision delivered, the view says it once, as its event
+// `delivered` with the revision, the group and the heads, which the issue tracker derives "fix delivered" from.
+func TestProcessWholeSaysTheSourceRevisionIsDelivered(t *testing.T) {
+	srv, err := server.NewServer(&server.Options{Host: "127.0.0.1", Port: -1, JetStream: true, StoreDir: t.TempDir(), NoLog: true, NoSigs: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	go srv.Start()
+	if !srv.ReadyForConnections(10 * time.Second) {
+		t.Fatal("the bus did not come up")
+	}
+	t.Cleanup(srv.Shutdown)
+	nc, err := nats.Connect(srv.ClientURL())
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(nc.Close)
+	js, err := jetstream.New(nc)
+	if err != nil {
+		t.Fatal(err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+	defer cancel()
+	events, err := js.CreateStream(ctx, jetstream.StreamConfig{Name: "VIEWEVENTS", Subjects: []string{"mesh.mod.mesh-delivery-view.event.>"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	kvs := map[string]jetstream.KeyValue{}
+	for _, name := range []string{delivery.SourceBucket, delivery.ModulesBucket, delivery.AssignerWantsBucket, delivery.HaveBucket} {
+		kv, err := js.CreateKeyValue(ctx, jetstream.KeyValueConfig{Bucket: name, History: 5})
+		if err != nil {
+			t.Fatal(err)
+		}
+		kvs[name] = kv
+	}
+	dir, err := os.MkdirTemp(os.Getenv("TMPDIR"), "i")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { os.RemoveAll(dir) })
+	broker := filepath.Join(dir, "broker")
+	cred, _ := json.Marshal(credential{URL: srv.ClientURL(), Node: "n", Module: "mesh-delivery-view", User: "mesh-delivery-view", Password: "x"})
+	if err := os.WriteFile(broker, cred, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	env := map[string]string{"MESH_BROKER_FILE": broker, "MESH_DELIVERY_SOCKET": filepath.Join(dir, "run", "cli.sock"), "MESH_DELIVERY_WEB": "127.0.0.1:0"}
+	runCtx, stop := context.WithCancel(ctx)
+	done := make(chan error, 1)
+	go func() { done <- run(runCtx, func(k string) string { return env[k] }, io.Discard) }()
+	t.Cleanup(func() { stop(); <-done })
+
+	fp := "src2:" + strings.Repeat("2", 64)
+	mod, _ := delivery.EncodeModule(delivery.Module{FP2: fp, Manifest: strings.Repeat("a", 64), Repository: "novox/web", Commit: strings.Repeat("1", 40), Path: delivery.RootPath})
+	want, _ := delivery.EncodeWant(delivery.Want{FP: fp, Composed: "sha256:" + strings.Repeat("c", 64)})
+	failed, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthFailed})
+	good, _ := delivery.EncodeHave(delivery.Have{FP: fp, Phase: delivery.PhaseRunning, Health: delivery.HealthOK})
+	src, _ := delivery.EncodeSource(delivery.Source{Group: "feat/x", Heads: map[string]string{"novox/web": strings.Repeat("1", 40)}})
+	for bucket, kv := range map[string][]byte{delivery.ModulesBucket: mod} {
+		if _, err := kvs[bucket].Put(ctx, "web", kv); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if _, err := kvs[delivery.AssignerWantsBucket].Put(ctx, "want.ace.web", want); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := kvs[delivery.HaveBucket].Put(ctx, "ace.web", failed); err != nil {
+		t.Fatal(err)
+	}
+	rev, err := kvs[delivery.SourceBucket].Put(ctx, delivery.SourceMainKey, src)
+	if err != nil {
+		t.Fatal(err)
+	}
+	time.Sleep(time.Second)
+	if info, err := events.Info(ctx); err != nil || info.State.Msgs != 0 {
+		t.Fatalf("an event was said before the revision was delivered: %v %+v", err, info.State)
+	}
+	if _, err := kvs[delivery.HaveBucket].Put(ctx, "ace.web", good); err != nil {
+		t.Fatal(err)
+	}
+	var msg *jetstream.RawStreamMsg
+	for deadline := time.Now().Add(20 * time.Second); time.Now().Before(deadline) && msg == nil; time.Sleep(100 * time.Millisecond) {
+		msg, _ = events.GetLastMsgForSubject(ctx, "mesh.mod.mesh-delivery-view.event."+DeliveredEventType)
+	}
+	if msg == nil {
+		t.Fatal("the delivery of the source revision was never said")
+	}
+	var e view.DeliveredEvent
+	if err := json.Unmarshal(msg.Data, &e); err != nil || e.Revision != rev || e.Group != "feat/x" || e.Heads["novox/web"] != strings.Repeat("1", 40) {
+		t.Errorf("the event = %+v, %v; want revision %d, group feat/x and the heads", e, err, rev)
+	}
+	if msg.Header.Get("x-source") != "mesh-delivery-view" || msg.Header.Get("x-event-id") == "" {
+		t.Errorf("the headers = %v", msg.Header)
+	}
+}

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

@@ -138,6 +138,11 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		return err
 	}
 	model.SetWriter(kvWriter{ctx: ctx, js: js})
+	model.SetDeliveredEvent(func(e view.DeliveredEvent) {
+		if err := emitEvent(ctx, js, cred, DeliveredEventType, e, fmt.Sprintf("delivered.%d", e.Revision)); err != nil {
+			logger.Printf("the delivery of source revision %d could not be said: %v", e.Revision, err)
+		}
+	})
 	model.SetBacklog(view.BacklogOver(js, 5*time.Second))
 	for _, bucket := range []string{delivery.HaveBucket, view.AssignmentsBucket, view.BuildsBucket, delivery.PresenceBucket, delivery.StuckBucket, delivery.CandidatesBucket, delivery.ModulesBucket, delivery.AsksBucket, delivery.AnswersBucket, delivery.BuilderBuildsBucket, delivery.AssignerWantsBucket, delivery.SourceBucket, delivery.ControllerComposedBucket, delivery.AssignerAsksBucket, delivery.ViewAnswersBucket} {
 		go follow(ctx, js, bucket, model, logger)
@@ -211,3 +216,28 @@ func watchOnce(ctx context.Context, js jetstream.JetStream, bucket string, model
 		}
 	}
 }
+
+// DeliveredEventType is the event the view says when a source revision is delivered: mesh.mod.mesh-delivery-view.event.delivered
+// (issue 1906 D9, ADR 0539 §17). Its body is view.DeliveredEvent.
+const DeliveredEventType = "delivered"
+
+// emitEvent publishes an event of the view's own namespace, mesh.mod.<module>.event.<type>, with the SDK's envelope headers
+// (as mesh-sdk's process.Conn.Emit does) and the id as the bus's own dedup id, so the same revision said twice in the stream's
+// duplicate window is kept once.
+func emitEvent(ctx context.Context, js jetstream.JetStream, cred credential, eventType string, body any, id string) error {
+	data, err := json.Marshal(body)
+	if err != nil {
+		return err
+	}
+	msg := nats.NewMsg("mesh.mod." + cred.Module + ".event." + eventType)
+	msg.Header.Set("x-event-id", id)
+	msg.Header.Set("x-source", cred.Module)
+	msg.Header.Set("x-node", cred.Node)
+	msg.Header.Set("x-time", time.Now().UTC().Format(time.RFC3339Nano))
+	msg.Header.Set("content-type", "application/json")
+	msg.Data = data
+	pub, cancel := context.WithTimeout(ctx, 10*time.Second)
+	defer cancel()
+	_, err = js.PublishMsg(pub, msg, jetstream.WithMsgID(id))
+	return err
+}

modules/mesh-delivery-view/internal/view/view.go

@@ -119,6 +119,8 @@ type SourceRow struct {
 	Revision uint64   `json:"revision"`
 	State    string   `json:"state"`
 	Waiting  []string `json:"waiting,omitempty"`
+	// Wants is how many wants of modules at the indexer were looked at; a revision with none is delivered of nothing.
+	Wants int `json:"wants"`
 }
 
 // AskRow is an open ask (novox/hq ADR 0437 §10): the gate's ask with no answer under its key in the teller's answers.
@@ -210,24 +212,26 @@ type UnreadableRow struct {
 
 // Model holds the rows' inputs in memory and nothing else: a restart fills it again from the buckets' last values.
 type Model struct {
-	mu       sync.Mutex
-	assigns  map[string]assignment
-	builds   map[string]buildEntry          // by <module>.build.<id>
-	groups   map[string]groupEntry          // the indexer's candidates, by group
-	modules  map[string]moduleEntry         // the indexer's modules, by module
-	keys     map[string]keyEntry            // the builder's build keys, by <module>.<fingerprint>
-	asks     map[string]askRecord           // the gate's asks, by key
-	wants    map[string]wantEntry           // the assigner's wants, by <node>.<module>
-	source   *sourceEntry                   // the gate's source key
-	composed map[string]composedEntry       // the controller's digests and judgements, by <node>.<module>.<hex>
-	aasks    map[string]assignerAskRecord   // the assigner's asks (hold, restore), by key
-	chosen   map[string]delivery.ViewAnswer // the answers the view wrote, by ask key
-	writer   Writer
-	backlog  *Backlog
-	answers  map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
-	newest   map[string]string // module -> id
-	haves    map[string]haveEntry
-	nodes    map[string]nodeEntry
+	mu             sync.Mutex
+	assigns        map[string]assignment
+	builds         map[string]buildEntry  // by <module>.build.<id>
+	groups         map[string]groupEntry  // the indexer's candidates, by group
+	modules        map[string]moduleEntry // the indexer's modules, by module
+	keys           map[string]keyEntry    // the builder's build keys, by <module>.<fingerprint>
+	asks           map[string]askRecord   // the gate's asks, by key
+	wants          map[string]wantEntry   // the assigner's wants, by <node>.<module>
+	source         *sourceEntry           // the gate's source key
+	deliveredEvent func(DeliveredEvent)
+	said           uint64                         // the source revision whose delivery was said
+	composed       map[string]composedEntry       // the controller's digests and judgements, by <node>.<module>.<hex>
+	aasks          map[string]assignerAskRecord   // the assigner's asks (hold, restore), by key
+	chosen         map[string]delivery.ViewAnswer // the answers the view wrote, by ask key
+	writer         Writer
+	backlog        *Backlog
+	answers        map[string]string // by key: the router id of the asking the teller's answer is for ("" when unreadable)
+	newest         map[string]string // module -> id
+	haves          map[string]haveEntry
+	nodes          map[string]nodeEntry
 	// the watchman's buckets, by key
 	presence map[string]presenceEntry
 	stuck    map[string]stuckEntry
@@ -302,6 +306,8 @@ type composedEntry struct {
 
 type sourceEntry struct {
 	revision uint64
+	group    string
+	heads    map[string]string
 	err      string
 }
 
@@ -339,10 +345,50 @@ func New() *Model {
 		presence: map[string]presenceEntry{}, stuck: map[string]stuckEntry{}}
 }
 
-// Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed.
+// Apply takes one change of a key. A value it cannot read is kept as an error on its row, never guessed. When the change makes
+// the source revision delivered, the event of it is said once (SetDeliveredEvent), after the model's lock is released.
 func (m *Model) Apply(e Entry) {
+	m.mu.Lock()
+	m.apply(e)
+	event := m.deliveredNow()
+	say := m.deliveredEvent
+	m.mu.Unlock()
+	if event != nil && say != nil {
+		say(*event)
+	}
+}
+
+// DeliveredEvent is what the view says when the source revision is delivered (event `delivered` of the view, issue 1906 D9, ADR
+// 0539 §17): the revision of the gate's source key, the group that moved it and every repository's merged head at it. A
+// consumer takes every earlier revision as covered by it, and takes the same revision twice as once (after a restart of the view).
+type DeliveredEvent struct {
+	Revision uint64            `json:"revision"`
+	Group    string            `json:"group"`
+	Heads    map[string]string `json:"heads"`
+}
+
+// SetDeliveredEvent sets where the event of a delivered source revision goes; nothing is said until it is set.
+func (m *Model) SetDeliveredEvent(say func(DeliveredEvent)) {
 	m.mu.Lock()
 	defer m.mu.Unlock()
+	m.deliveredEvent = say
+}
+
+// deliveredNow is the event of the source revision when it is delivered now and has not been said, or nil. A revision that went
+// bad after it was delivered and recovered is not said again.
+func (m *Model) deliveredNow() *DeliveredEvent {
+	if m.deliveredEvent == nil || m.source == nil || m.source.err != "" || m.said == m.source.revision {
+		return nil
+	}
+	row := m.sourceRow()
+	if row == nil || row.State != SourceDelivered || row.Wants == 0 {
+		return nil
+	}
+	m.said = m.source.revision
+	return &DeliveredEvent{Revision: m.source.revision, Group: m.source.group, Heads: m.source.heads}
+}
+
+func (m *Model) apply(e Entry) {
 	switch e.Bucket {
 	case delivery.HaveBucket:
 		node, module, isNode, err := delivery.ParseHaveKey(e.Key)
@@ -456,8 +502,8 @@ func (m *Model) Apply(e Entry) {
 			m.source = nil
 			return
 		}
-		_, err := delivery.DecodeSource(e.Value)
-		m.source = &sourceEntry{revision: e.Revision, err: errText(err)}
+		src, err := delivery.DecodeSource(e.Value)
+		m.source = &sourceEntry{revision: e.Revision, group: src.Group, heads: src.Heads, err: errText(err)}
 	case delivery.AnswersBucket:
 		if e.Deleted {
 			delete(m.answers, e.Key)
@@ -963,12 +1009,14 @@ func (m *Model) sourceRow() *SourceRow {
 		}
 		return fp
 	}
+	looked := 0
 	for key, w := range m.wants {
 		node, module, _ := strings.Cut(key, ".")
 		mod, ok := m.modules[module]
 		if w.err != "" || w.value.Removed || !ok || mod.err != "" || mod.value.FP2 == "" {
 			continue
 		}
+		looked++
 		h := m.haves[key]
 		switch {
 		case w.value.FP != mod.value.FP2:
@@ -993,7 +1041,7 @@ func (m *Model) sourceRow() *SourceRow {
 		}
 	}
 	sort.Strings(waiting)
-	row := &SourceRow{Revision: m.source.revision, State: SourceDelivered}
+	row := &SourceRow{Revision: m.source.revision, State: SourceDelivered, Wants: looked}
 	if len(waiting) > 0 {
 		row.State, row.Waiting = SourceNotDelivered, waiting
 	}

modules/mesh-delivery-view/internal/view/wants_test.go

@@ -156,3 +156,37 @@ func TestTheJudgementOfTheMoveIsShownBesideTheWant(t *testing.T) {
 		t.Errorf("text = %v", text)
 	}
 }
+
+// 1906/D9 (asked by the lead for issue 1946): the view says, once for each source revision, that it is delivered: the event the
+// issue tracker derives "fix delivered" from, in place of the gate's states that go in release 6 (ADR 0539 §17).
+func TestTheViewSaysOnceThatASourceRevisionIsDelivered(t *testing.T) {
+	m := New()
+	var said []DeliveredEvent
+	m.SetDeliveredEvent(func(e DeliveredEvent) { said = append(said, e) })
+	fill(m, sourceE(t, 14355), moduleE(t, "web", fpNew), assign("ace", "web", "b-1", ""),
+		wantE(t, "ace", "web", delivery.Want{FP: fpNew}))
+	if len(said) != 0 {
+		t.Fatalf("said %v before any have", said)
+	}
+	fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthFailed))
+	if len(said) != 0 {
+		t.Fatalf("said %v for a failed have", said)
+	}
+	fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK))
+	if len(said) != 1 || said[0].Revision != 14355 || said[0].Group != "g" || said[0].Heads["novox/web"] == "" {
+		t.Fatalf("said = %+v, want revision 14355 once with its group and heads", said)
+	}
+	fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthOK), wantE(t, "ace", "web", delivery.Want{FP: fpNew}))
+	if len(said) != 1 {
+		t.Errorf("said again for the same revision: %v", said)
+	}
+	// a newer revision is delivered when its wants and haves are; a revision that goes bad and good again is not said twice
+	fill(m, sourceE(t, 14356))
+	if len(said) != 2 || said[1].Revision != 14356 {
+		t.Errorf("the next revision: %v", said)
+	}
+	fill(m, haveFP(t, "ace", "web", fpNew, delivery.HealthFailed), haveFP(t, "ace", "web", fpNew, delivery.HealthOK))
+	if len(said) != 2 {
+		t.Errorf("said again after a have failed and recovered: %v", said)
+	}
+}

modules/mesh-delivery-view/manifest_test.go

@@ -89,11 +89,21 @@ func TestManifestIsTakenAndReadsOnly(t *testing.T) { // 1134/AC7
 	if !reflect.DeepEqual(got, want) {
 		t.Errorf("reads = %v, want exactly %v", got, want)
 	}
-	for _, k := range []string{"emits", "consumes", "provides", "claims"} {
+	for _, k := range []string{"consumes", "provides", "claims"} {
 		if _, ok := top[k]; ok {
 			t.Errorf("the manifest declares %q: the view writes only its answers", k)
 		}
 	}
+	// The one event the view says: a source revision is delivered (1906/D9, ADR 0539 §17), derived by it from the wants and the haves.
+	var emits []string
+	if raw, ok := top["emits"]; ok {
+		if err := json.Unmarshal(raw, &emits); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if !reflect.DeepEqual(emits, []string{"delivered"}) {
+		t.Errorf("emits = %v, want exactly [delivered]", emits)
+	}
 	// The view uses no build seat: it reads their workers by the broker's named grant and publishes no ask (ADR 0525 §2).
 	if len(m.Uses) != 0 {
 		t.Errorf("uses = %v, want none", m.Uses)

modules/mesh-delivery-view/module.json

@@ -15,6 +15,9 @@
     "assigner_differences",
     "node_haves"
   ],
+  "emits": [
+    "delivered"
+  ],
   "state": [
     { "name": "answers", "history": 5 }
   ],