diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 67cadbe5..accfbcfb 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -75,9 +75,34 @@ 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 a finding 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 after the grace, not at once", 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) + } + expire(w, c, signalKey("take-in", revKey), TakeInReplayGrace) }, 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))) + 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.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 { + t.Fatalf("findings past the bound: %v", o.findings) + } + w.TakeIn(takeInEvent(t, delivery.TakeInTaken, revKey), takeInBody(t, delivery.TakeInTaken, revKey, t0)) + if len(o.findings) != 0 { + t.Fatalf("the drop left %v", o.findings) + } + 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 { b, err := delivery.EncodeModule(delivery.Module{FP2: "src2:" + strings.Repeat("a", 64), Manifest: strings.Repeat("b", 64), diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index 8d9abf23..d0bfb861 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/modules/mesh-delivery-watchman/internal/watch/takein.go @@ -19,6 +19,11 @@ 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" @@ -61,9 +66,7 @@ func (w *Watchman) TakeIn(event string, body []byte) { 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 { - w.unsignal(takeInRow, key) - w.find(f) - return + left = TakeInReplayGrace } w.unfind(finding) w.signal(takeInRow, key, left, signalDue{Do: dueFind, Finding: f})