From ed4f07b0d8cf8eaae4c665fd90b2ff6b22180d0c Mon Sep 17 00:00:00 2001 From: jochen Date: Wed, 4 Nov 2026 23:10:23 +0100 Subject: [PATCH] =?UTF-8?q?Watchman:=20a=20take-in=20wait=20past=20its=20b?= =?UTF-8?q?ound=20heard=20in=20the=20replay=20is=20said=20once=20the=20con?= =?UTF-8?q?sumer=20has=20caught=20up,=20not=20after=20a=20time=20(issue=20?= =?UTF-8?q?1585,=20ADR=200633=20=C2=A75)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn --- .../cmd/mesh-delivery-watchman/main.go | 12 ++++++ .../internal/watch/findings_test.go | 14 ++++--- .../internal/watch/takein.go | 37 ++++++++++++++----- .../internal/watch/watch.go | 5 ++- 4 files changed, 53 insertions(+), 15 deletions(-) diff --git a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go index 8117b1d9..239ad6e4 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/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) } diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index accfbcfb..0454d76f 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/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. 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 { diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index d0bfb861..9cff07b2 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/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{} } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index 67b641fe..e8bd18b9 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/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{},