modules/mesh-delivery-indexer/internal/index/pending.go

@@ -1086,13 +1086,47 @@ func (ix *Indexer) takeInWholeBy(ctx context.Context, entries []PendingEntry, ow
 	}
 	ix.forget(claimed...)
 	ix.logf("take-in: %d merges taken in as one revision by %s%s, oldest %s", len(claimed), own, byWords(by), claimed[0].Key)
-	for _, t := range saidWaits(claimed) {
+	for _, t := range takenWaits(claimed) {
 		t.By, t.Why = who, why
 		ix.sayTakeIn(ctx, t)
 	}
 	return applied, nil
 }
 
+// takenWaits are the taken events of a take-in: every wait the entries were said under (saidWaits), every wait
+// indexer_pending names the claimed entries by now (WaitsOf: a revision by its oldest merge's key, a group by its g. key) and
+// the key of every claimed entry, since indexer_pending named it by that key earlier when it waited behind a group. So a wait
+// the watchman learned of without the indexer marking it (seeded from indexer_pending, or said by the own-build-unknown
+// path) ends too (design 47, Taken in by the indexer it names). No key is said twice.
+func takenWaits(entries []PendingEntry) []takein.TakeIn {
+	out := saidWaits(entries)
+	said := map[string]bool{}
+	for _, t := range out {
+		said[t.Key] = true
+	}
+	w := WaitsOf(entries)
+	var named []takein.TakeIn
+	if len(w.Ready) > 0 {
+		named = append(named, revisionWait(takein.TakeInTaken, w.Ready, w.Ready[len(w.Ready)-1].Value.WaitsFor))
+	}
+	for _, g := range slices.Sorted(maps.Keys(w.Groups)) {
+		named = append(named, groupWait(takein.TakeInTaken, g, w.Groups[g]))
+	}
+	for _, g := range slices.Sorted(maps.Keys(w.Behind)) {
+		named = append(named, revisionWait(takein.TakeInTaken, w.Behind[g], w.Behind[g][len(w.Behind[g])-1].Value.WaitsFor))
+	}
+	for _, e := range entries {
+		named = append(named, revisionWait(takein.TakeInTaken, []PendingEntry{e}, e.Value.WaitsFor))
+	}
+	for _, t := range named {
+		if !said[t.Key] {
+			said[t.Key] = true
+			out = append(out, t)
+		}
+	}
+	return out
+}
+
 func byWords(by string) string {
 	if by == "" {
 		return ""

modules/mesh-delivery-indexer/internal/index/pending_test.go

@@ -205,6 +205,33 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 			t.Errorf("no pending state: err %v, avahi at %s: want the merge taken in at once", err, b.commit(t, "avahi"))
 		}
 	})
+	t.Run("1966/AC195-2 AC195-3 a merge said pending with its build unknown and taken in on redelivery ends that wait under its key; a revision wait is said once", func(t *testing.T) {
+		b := newTakeInBed(t)
+		ownBuild := b.ix.OwnBuild
+		b.ix.ownFP, b.ix.ownFixed = "", false
+		b.ix.OwnBuild = func(context.Context) (string, error) { return "", errors.New("the engine does not answer") }
+		b.ix.FixOwnBuild(ctx)
+		b.said = nil
+		if _, _, err := b.ix.Arrive(ctx, 7, b.merge(30, sha('b'), "tree-avahi-2", "")); !errors.Is(err, ErrOwnBuildUnknown) {
+			t.Fatal(err)
+		}
+		b.ix.OwnBuild, b.ix.ownFP, b.ix.ownFixed = ownBuild, "", false
+		b.ix.FixOwnBuild(ctx)
+		b.arrive(t, 7, b.merge(30, sha('b'), "tree-avahi-2", "")) // the stream delivers it again
+		if want := []string{"pending " + takein.PendingKey(7), "taken " + takein.PendingKey(7)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 {
+			t.Fatalf("said %v, pending %v: want %v and nothing pending", b.events(), b.keys(), want)
+		}
+		// Two merges said pending under the oldest's key (waiting for a named indexer), then taken in together: one taken, under
+		// that key, and no entry's key said twice.
+		b.said = nil
+		b.arrive(t, 8, b.merge(31, sha('c'), "tree-avahi-3", "tree-indexer-2"))
+		b.arrive(t, 9, b.merge(32, sha('d'), "tree-avahi-4", "tree-indexer-2"))
+		b.said = nil
+		b.arrive(t, 10, b.merge(33, sha('e'), "tree-avahi-5", b.base.Trees["modules/"+Module]))
+		if want := []string{"taken " + takein.PendingKey(8), "taken " + takein.PendingKey(9), "taken " + takein.PendingKey(10)}; !slices.Equal(b.events(), want) || len(b.keys()) != 0 {
+			t.Errorf("said %v, pending %v: want %v, the wait's key (m.8, said pending) once and each other entry's own", b.events(), b.keys(), want)
+		}
+	})
 	t.Run("T7 and T3 a merge alone, an indexer change, a merge behind it and a revert", func(t *testing.T) {
 		b := newTakeInBed(t)
 		own, avahi0 := b.fp(t, Module), b.fp(t, "avahi")
@@ -225,9 +252,10 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 			t.Fatalf("taken in at the bind: %d verdicts (were %d), want the new manifest's", n, judgedBefore)
 		}
 		// A merge with no change to the indexer and nothing pending is taken in alone, as before.
-		if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || len(b.said) != 0 {
-			t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in", avahi1, avahi0, b.keys(), b.events())
+		if avahi1 := b.fp(t, "avahi"); avahi1 == avahi0 || len(b.keys()) != 0 || !slices.Equal(b.events(), []string{"taken " + takein.PendingKey(1)}) {
+			t.Fatalf("alone: avahi %s (was %s), pending %v, said %v: want taken in, said taken under its key (1966/AC195-1)", avahi1, avahi0, b.keys(), b.events())
 		}
+		b.said = nil
 		avahi1, verdicts := b.fp(t, "avahi"), len(b.ix.Verdicts.(*memStore).vals)
 		// T4: a change to the indexer writes only its entry (with its graph key and verdict); the rest is pending.
 		b.arrive(t, 2, b.merge(21, sha('c'), "tree-avahi-3", "tree-indexer-2"))
@@ -259,7 +287,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 		if len(b.keys()) != 0 || b.fp(t, Module) != own || b.commit(t, "avahi") != sha('e') {
 			t.Fatalf("revert: pending %v, indexer %s, avahi at %s: want all taken in", b.keys(), b.fp(t, Module), b.commit(t, "avahi"))
 		}
-		if want := []string{"pending " + takein.PendingKey(2), "taken " + takein.PendingKey(2)}; !slices.Equal(b.events(), want) {
+		if want := []string{"pending " + takein.PendingKey(2), "taken " + takein.PendingKey(2), "taken " + takein.PendingKey(3), "taken " + takein.PendingKey(4)}; !slices.Equal(b.events(), want) {
 			t.Errorf("said %v, want %v", b.events(), want)
 		}
 	})
@@ -459,7 +487,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 		if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') {
 			t.Fatalf("end: pending %v, avahi at %s: want the group and the merge behind it taken in", b.keys(), b.commit(t, "avahi"))
 		}
-		if want := []string{"pending g.feat/take-in", "pending g.feat/take-in", "taken g.feat/take-in"}; !slices.Equal(b.events(), want) {
+		if want := []string{"pending g.feat/take-in", "taken " + takein.PendingKey(2), "pending g.feat/take-in", "taken g.feat/take-in", "taken " + takein.PendingKey(1), "taken " + takein.PendingKey(3), "taken " + takein.PendingKey(4)}; !slices.Equal(b.events(), want) {
 			t.Errorf("said %v, want %v", b.events(), want)
 		}
 		// R1: the end published again after the take-in changes nothing and is not delivered again.
@@ -510,7 +538,7 @@ func TestTakeIn(t *testing.T) { // 1585/T3-T13 (ADR 0633)
 		if err := b.ix.PendingAgain(ctx); err != nil {
 			t.Fatal(err)
 		}
-		if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') || !slices.Equal(b.events(), []string{"taken g.feat/take-in"}) {
+		if len(b.keys()) != 0 || b.commit(t, "avahi") != sha('6') || !slices.Equal(b.events(), []string{"taken g.feat/take-in", "taken " + takein.PendingKey(2)}) {
 			t.Errorf("all taken: pending %v, avahi at %s, said %v: want the group ended and the merge behind it taken in", b.keys(), b.commit(t, "avahi"), b.events())
 		}
 		held := func() *takeInBed {

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

@@ -13,6 +13,8 @@ import (
 	"git.novox.be/novox/mesh-sdk/go/process"
 	"git.novox.be/novox/mesh-sdk/go/views"
 
+	"git.novox.be/novox/mesh-delivery/modules/mesh-delivery-indexer/takein"
+
 	"mesh-delivery-watchman/internal/watch"
 
 	"github.com/nats-io/nats.go/jetstream"
@@ -77,6 +79,34 @@ func TestRunNeedsItsCredential(t *testing.T) { // 1135/AC6
 	if err == nil || !strings.Contains(err.Error(), "MESH_BROKER_FILE") {
 		t.Fatalf("run = %v, want a refusal naming MESH_BROKER_FILE", err)
 	}
+	t.Run("1966/AC195-5: a seed indexer_pending could not answer, or answered unreadably, ends nothing", func(t *testing.T) {
+		for name, call := range map[string]func(string, any) (json.RawMessage, error){
+			"cannot be asked": func(string, any) (json.RawMessage, error) { return nil, errors.New("no answer") },
+			"unreadable":      func(string, any) (json.RawMessage, error) { return json.RawMessage(`{"waits":`), nil },
+		} {
+			out := &findingsOut{findings: map[string]bool{}}
+			w := watch.New(out, time.Now)
+			const key = "m.00000000000000000042"
+			old := watch.Finding{Scope: watch.ScopeDelivery, ID: key, Kind: watch.KindTakeInWaits, Severity: watch.Warning, Source: "take-in",
+				Since: time.Now().Add(-3 * time.Hour), Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"}
+			value, err := watch.EncodeFinding(old)
+			if err != nil {
+				t.Fatal(err)
+			}
+			out.findings[old.Key()] = true
+			w.SeedFinding(old.Key(), value)
+			body, err := takein.EncodeTakeIn(takein.TakeIn{Kind: takein.TakeInPending, Key: "m.00000000000000000043", What: takein.TakeInRevision, Since: time.Now(), WaitsFor: "src2:x",
+				Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}})
+			if err != nil {
+				t.Fatal(err)
+			}
+			w.TakeIn(takein.TakeInPendingEvent, body)
+			seedTakeIn(w, call, func(string, ...any) {})
+			if !out.findings[old.Key()] || out.pulseDeletes != 0 {
+				t.Errorf("%s: finding stands %v, signals ended %d: want the previous run's finding and the held wait kept", name, out.findings[old.Key()], out.pulseDeletes)
+			}
+		}
+	})
 	t.Run("A2j: a want stood since the assigner's bus wrote it, through the first view", func(t *testing.T) {
 		// The watchman's want wait starts at the entry's time (createdOr): through mesh-sdk's views.ProcessReader it is
 		// the time the owner's bus wrote the key, never the watchman's start.
@@ -113,11 +143,14 @@ type refusingBus struct{ err error }
 func (r refusingBus) KeyValue(context.Context, string) (jetstream.KeyValue, error) { return nil, r.err }
 
 // findingsOut keeps which findings stand.
-type findingsOut struct{ findings map[string]bool }
+type findingsOut struct {
+	findings     map[string]bool
+	pulseDeletes int
+}
 
 func (o *findingsOut) PutPresence(string, bool) error               { return nil }
 func (o *findingsOut) PutPulse(string, []byte, time.Duration) error { return nil }
-func (o *findingsOut) DeletePulse(string) error                     { return nil }
+func (o *findingsOut) DeletePulse(string) error                     { o.pulseDeletes++; return nil }
 func (o *findingsOut) PutStuck(string, []byte) error                { return nil }
 func (o *findingsOut) PutFinding(key string, _ []byte) error        { o.findings[key] = true; return nil }
 func (o *findingsOut) DeleteFinding(key string) error               { delete(o.findings, key); return nil }

modules/mesh-delivery-watchman/internal/watch/findings_test.go

@@ -105,6 +105,36 @@ func TestTheWatchmansFindings(t *testing.T) {
 			}
 			w.TakeInCaughtUp()
 		}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
+		{"1966/AC195-5: a seed ends the finding and the held wait of a key indexer_pending no longer names, and keeps one it names with its own since", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			// A previous run's finding for revKey stands in the bucket (read at the start); a wait of staleKey is held from an event.
+			old := Finding{Scope: ScopeDelivery, ID: revKey, Kind: KindTakeInWaits, Severity: Warning, Source: "take-in", Since: t0.Add(-3 * time.Hour), Observer: Observer, Summary: "held", Said: "held", Headline: "held", Explanation: "held", Needs: "a take-in"}
+			value, err := EncodeFinding(old)
+			if err != nil {
+				t.Fatal(err)
+			}
+			o.findings[old.Key()] = old
+			w.SeedFinding(old.Key(), value)
+			const staleKey = "m.00000000000000000099"
+			w.TakeInCaughtUp()
+			w.TakeIn(takeInEvent(t, takein.TakeInPending, staleKey), takeInBody(t, takein.TakeInPending, staleKey, c.t))
+			w.SeedTakeIn(takein.PendingAnswer{Waits: []takein.PendingWait{{Key: groupKey, What: takein.TakeInGroup, Keys: []string{groupKey},
+				Merges: []takein.Merge{{Repository: "novox/a", Number: 1, Merge: "c0ffee"}}, Since: c.t.Add(-10 * time.Minute)}}})
+			if _, ok := o.findings[old.Key()]; ok || w.signals[signalKey("take-in", staleKey)].Finding.ID != "" || w.signals[signalKey("take-in", revKey)].Finding.ID != "" {
+				t.Fatalf("a wait the answer does not name stands: findings %v, signals %v", o.findings, w.signals)
+			}
+			if o.ttls[signalKey("take-in", groupKey)] != TakeInGroupBound-10*time.Minute {
+				t.Fatalf("the named wait: ttl %v, want the rest of its bound from its own since", o.ttls[signalKey("take-in", groupKey)])
+			}
+		}, map[string]string{}},
+		{"1966/AC195-4: a taken event for a key the watchman holds no wait for changes no finding and no signal", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			w.TakeInCaughtUp()
+			w.TakeIn(takeInEvent(t, takein.TakeInPending, groupKey), takeInBody(t, takein.TakeInPending, groupKey, t0))
+			ttls, pulses := len(o.ttls), len(o.pulses)
+			w.TakeIn(takeInEvent(t, takein.TakeInTaken, revKey), takeInBody(t, takein.TakeInTaken, revKey, t0))
+			if o.findingWrites != 0 || len(o.ttls) != ttls || len(o.pulses) != pulses || o.ttls[signalKey("take-in", groupKey)] != TakeInGroupBound {
+				t.Fatalf("a taken for an unheld key: findings written %d, ttls %v, pulses %v", o.findingWrites, o.ttls, o.pulses)
+			}
+		}, map[string]string{}},
 		{"ADR 0633 §5 (1585/T21): a replayed pending followed by its taken never raises", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
 			w.TakeIn(takeInEvent(t, takein.TakeInPending, revKey), takeInBody(t, takein.TakeInPending, revKey, t0.Add(-2*time.Hour)))
 			w.TakeIn(takeInEvent(t, takein.TakeInTaken, revKey), takeInBody(t, takein.TakeInTaken, revKey, t0.Add(-2*time.Hour)))

modules/mesh-delivery-watchman/internal/watch/takein.go

@@ -52,10 +52,36 @@ func (w *Watchman) TakeIn(event string, body []byte) {
 // and group the indexer holds now, as indexer_pending answers them (ADR 0633 §5): the consumer is durable, so the
 // pending events it acknowledged before a restart are not heard again, and a wait within its bound would otherwise be
 // lost. Each wait's bound runs from its own since, as a pending event's does; the stream's events that follow replace
-// or end it.
+// or end it. A wait or finding the answer does not name is ended.
 func (w *Watchman) SeedTakeIn(a takein.PendingAnswer) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
+	// A wait or finding of a key the answer does not name is ended as a take-in.taken would end it: the finding a previous
+	// run wrote (read at the start, before the seed), a wait held from an event, a signal standing (design 47, Taken in by
+	// the indexer it names). Only an answered seed gets here: one that could not ask ends nothing.
+	named := map[string]bool{}
+	for _, wait := range a.Waits {
+		named[wait.Key] = true
+	}
+	stale := map[string]bool{}
+	for _, value := range w.findings {
+		if f, err := DecodeFinding(value); err == nil && f.Scope == ScopeDelivery && f.Kind == KindTakeInWaits && !named[f.ID] {
+			stale[f.ID] = true
+		}
+	}
+	for _, due := range w.signals {
+		if due.Finding.Kind == KindTakeInWaits && due.Finding.Scope == ScopeDelivery && !named[due.Finding.ID] {
+			stale[due.Finding.ID] = true
+		}
+	}
+	for key := range w.takeInHeld {
+		if !named[key] {
+			stale[key] = true
+		}
+	}
+	for _, key := range slices.Sorted(maps.Keys(stale)) {
+		w.takeIn(takein.TakeInTaken, key, takein.TakeIn{Kind: takein.TakeInTaken, Key: key})
+	}
 	for _, wait := range a.Waits {
 		w.takeIn(takein.TakeInPending, wait.Key, takein.TakeIn{Kind: takein.TakeInPending, Key: wait.Key, What: wait.What,
 			Merges: wait.Merges, Group: wait.Group, Members: wait.Members, WaitsFor: wait.WaitsFor, Since: wait.Since})