diff --git a/modules/mesh-delivery-watchman/internal/watch/findings_test.go b/modules/mesh-delivery-watchman/internal/watch/findings_test.go index 30457c62..2363ea72 100644 --- a/modules/mesh-delivery-watchman/internal/watch/findings_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/findings_test.go @@ -105,6 +105,27 @@ 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)) diff --git a/modules/mesh-delivery-watchman/internal/watch/takein.go b/modules/mesh-delivery-watchman/internal/watch/takein.go index 171aa403..cc6b6c2e 100644 --- a/modules/mesh-delivery-watchman/internal/watch/takein.go +++ b/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})