Watchman: a take-in wait past its bound heard in the replay is said once the consumer has caught up, not after a time (issue 1585, ADR 0633 §5)

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

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

@@ -384,6 +384,15 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro
 		logger.Printf("the consumer %s_%s on %s does not exist yet, so no take-in wait is said until the watchman starts again: %v",
 			cred.Node, cred.Module, catalogue.StreamBusName(delivery.TakeInStream), err)
 	default:
+		// The replay has caught up when a message says none is pending after it, or when the consumer had none pending at its
+		// bind (the stream holds nothing it has not acknowledged): the waits past their bound are said then, not before.
+		if info, err := tc.Info(ctx); err != nil || (info.NumPending == 0 && info.NumAckPending == 0) {
+			if err != nil {
+				logger.Printf("the take-in consumer cannot be read, so its replay is caught up at its last message: %v", err)
+			} else {
+				w.TakeInCaughtUp()
+			}
+		}
 		hc, err := process.HearEvents(tc, func(m jetstream.Msg) { takeInHeard(w, m, logger.Printf) }, logger.Printf)
 		if err != nil {
 			return err
@@ -464,6 +473,9 @@ func takeInHeard(w *watch.Watchman, m jetstream.Msg, logf func(string, ...any))
 	} else {
 		w.TakeIn(event, m.Data())
 	}
+	if meta, err := m.Metadata(); err == nil && meta.NumPending == 0 {
+		w.TakeInCaughtUp()
+	}
 	if err := m.Ack(); err != nil {
 		logf("a take-in event could not be acknowledged: %v", err)
 	}

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

@@ -35,6 +35,7 @@ func TestTheWatchmansFindings(t *testing.T) {
 		want map[string]string
 	}{
 		{"ADR 0633 §5 (1585/T21): a revision waiting for an indexer build is a finding past 60 minutes from its since, not within; a group past 30", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			w.TakeInCaughtUp() // a live message, heard after the replay
 			w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0))
 			if o.ttls[signalKey("take-in", revKey)] != TakeInRevisionBound || len(o.findings) != 0 {
 				t.Fatalf("a revision's wait: ttls %v, findings %v", o.ttls, o.findings)
@@ -50,6 +51,7 @@ func TestTheWatchmansFindings(t *testing.T) {
 			expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound-TakeInGroupBound)
 		}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning, FindingKey(ScopeDelivery, groupKey, KindTakeInWaits): Warning}},
 		{"ADR 0633 §5 (1585/T21): the take-in.taken of a named indexer, a hand take-in or a drop clears the finding", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			w.TakeInCaughtUp() // a live message, heard after the replay
 			for _, by := range []string{"", "operator@shanks"} {
 				w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0))
 				expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound)
@@ -66,6 +68,7 @@ func TestTheWatchmansFindings(t *testing.T) {
 			}
 		}, map[string]string{}},
 		{"ADR 0633 §5 (1585/T21): a group's take-in.taken ends its 30 minutes, and the revision then pending under m.<seq> starts its own 60", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			w.TakeInCaughtUp() // a live message, heard after the replay
 			w.TakeIn(takeInEvent(t, delivery.TakeInPending, groupKey), takeInBody(t, delivery.TakeInPending, groupKey, t0))
 			expire(w, c, signalKey("take-in", groupKey), 25*time.Minute)
 			w.TakeIn(takeInEvent(t, delivery.TakeInTaken, groupKey), takeInBody(t, delivery.TakeInTaken, groupKey, t0))
@@ -75,21 +78,23 @@ func TestTheWatchmansFindings(t *testing.T) {
 			}
 			expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound)
 		}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
-		{"ADR 0633 §5 (1585/T21): a pending event heard again at the start, since past its bound, is said after the grace, not at once", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+		{"ADR 0633 §5 (1585/T21): a pending event heard again at the start, since past its bound, is said once the replay has caught up, not before", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
 			w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0.Add(-2*time.Hour)))
-			if len(o.findings) != 0 || o.ttls[signalKey("take-in", revKey)] != TakeInReplayGrace {
-				t.Fatalf("a replayed old wait: findings %v, ttls %v", o.findings, o.ttls)
+			if len(o.findings) != 0 || len(o.pulses) != 0 {
+				t.Fatalf("a replayed old wait before the replay caught up: findings %v, pulses %v", o.findings, o.pulses)
 			}
-			expire(w, c, signalKey("take-in", revKey), TakeInReplayGrace)
+			w.TakeInCaughtUp()
 		}, map[string]string{FindingKey(ScopeDelivery, revKey, KindTakeInWaits): Warning}},
 		{"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, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0.Add(-2*time.Hour)))
 			w.TakeIn(takeInEvent(t, delivery.TakeInTaken, revKey), takeInBody(t, delivery.TakeInTaken, revKey, t0.Add(-2*time.Hour)))
+			w.TakeInCaughtUp()
 			if o.findingWrites != 0 || len(o.pulses) != 0 {
 				t.Fatalf("a wait that ended in the replay wrote %d findings and left pulses %v", o.findingWrites, o.pulses)
 			}
 		}, map[string]string{}},
 		{"ADR 0633 §5 (1585/T21): a drop clears the wait, and the merges left under it said pending again with the same since are raised from it", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
+			w.TakeInCaughtUp() // a live message, heard after the replay
 			w.TakeIn(takeInEvent(t, delivery.TakeInPending, revKey), takeInBody(t, delivery.TakeInPending, revKey, t0))
 			expire(w, c, signalKey("take-in", revKey), TakeInRevisionBound)
 			if len(o.findings) != 1 {
@@ -101,7 +106,6 @@ func TestTheWatchmansFindings(t *testing.T) {
 			}
 			next := "m.00000000000000000043"
 			w.TakeIn(takeInEvent(t, delivery.TakeInPending, next), takeInBody(t, delivery.TakeInPending, next, t0))
-			expire(w, c, signalKey("take-in", next), TakeInReplayGrace)
 		}, map[string]string{FindingKey(ScopeDelivery, "m.00000000000000000043", KindTakeInWaits): Warning}},
 		{"ADR 0578 §1: a module whose value carries a refusal of its manifest is a finding until it is accepted", func(t *testing.T, w *Watchman, o *fakeOut, c *clock) {
 			value := func(verdict string, problems ...string) []byte {

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

@@ -2,6 +2,8 @@ package watch
 
 import (
 	"fmt"
+	"maps"
+	"slices"
 	"strings"
 	"time"
 
@@ -19,11 +21,6 @@ const (
 	TakeInGroupBound    = 30 * time.Minute
 )
 
-// TakeInReplayGrace is how long a pending event whose bound has already passed waits before it is said: the stream keeps every
-// event, so a restarted watchman hears a wait's pending event and then, in the same pass, the taken event that ended it, and
-// that wait must never be raised. A wait still open after the grace is said.
-const TakeInReplayGrace = 30 * time.Second
-
 // takeInRow is the signals table's row of a take-in wait.
 const takeInRow = "take-in"
 
@@ -45,6 +42,7 @@ func (w *Watchman) TakeIn(event string, body []byte) {
 	defer w.mu.Unlock()
 	finding := FindingKey(ScopeDelivery, key, KindTakeInWaits)
 	if kind == delivery.TakeInTaken {
+		delete(w.takeInHeld, key)
 		w.unsignal(takeInRow, key)
 		w.unfind(finding)
 		return
@@ -65,9 +63,30 @@ func (w *Watchman) TakeIn(event string, body []byte) {
 			"not reach the machines.",
 		Needs: "the indexer build it waits for to run, or the take-in by hand (indexer_take_in_pending), or a drop (indexer_drop_pending)"}
 	left := bound - w.now().Sub(t.Since)
-	if left <= 0 {
-		left = TakeInReplayGrace
-	}
 	w.unfind(finding)
-	w.signal(takeInRow, key, left, signalDue{Do: dueFind, Finding: f})
+	w.unsignal(takeInRow, key)
+	delete(w.takeInHeld, key)
+	switch {
+	case left > 0:
+		w.signal(takeInRow, key, left, signalDue{Do: dueFind, Finding: f})
+	case !w.takeInCaught:
+		// Past its bound while the stream is still being replayed: the taken event that ended it may be a few messages on,
+		// so it is said once the replay has caught up (TakeInCaughtUp), and a taken event before that cancels it.
+		w.takeInHeld[key] = f
+	default:
+		w.find(f)
+	}
+}
+
+// TakeInCaughtUp tells the watchman its consumer has read the stream up to its end (the last message it delivered said no
+// message pending, or the consumer had none pending at its bind): every wait past its bound that no taken event ended is
+// said now, and the ones heard after are said as they come.
+func (w *Watchman) TakeInCaughtUp() {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.takeInCaught = true
+	for _, key := range slices.Sorted(maps.Keys(w.takeInHeld)) {
+		w.find(w.takeInHeld[key])
+	}
+	w.takeInHeld = map[string]Finding{}
 }

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

@@ -74,6 +74,9 @@ type Watchman struct {
 	seededFindings map[string]bool
 	// signals is every signal's pulse standing, by pulse key, with what its expiry does.
 	signals map[string]signalDue
+	// takeInHeld are the take-in waits past their bound heard while the stream replays, said once it has caught up (takein.go).
+	takeInHeld   map[string]Finding
+	takeInCaught bool
 	// away is each machine that said it sleeps or shuts down, until it says it is back.
 	away map[string]bool
 	// haveState is each module's have as last read, unhealthy its writes in a row that said it failed, advised each
@@ -107,7 +110,7 @@ type Watchman struct {
 func New(out Out, now func() time.Time) *Watchman {
 	return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{},
 		pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, wants: map[string]string{}, haves: map[string]string{}, held: map[string]bool{}, starts: map[string]startMark{}, bound: delivery.PulseMoveBound, stepBound: StepBound,
-		findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, away: map[string]bool{},
+		findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, takeInHeld: map[string]Finding{}, away: map[string]bool{},
 		haveState: map[string]delivery.Have{}, unhealthy: map[string]int{}, advised: map[string]int{}, advisedSince: map[string]time.Time{},
 		control: map[string]bool{}, failedSince: map[string]time.Time{}, provFailed: map[string]map[string]ProvReport{}, closures: map[string]map[string]time.Time{}, waiting: map[string]map[string]string{},
 		health: map[string]delivery.NodeHealth{}, tunnels: map[string]string{},