The watchman's seed ends the waits and findings indexer_pending no longer names (issue 1966, note 195)
SeedTakeIn, run once indexer_pending answered and after the findings of a previous run were read, ends every take-in wait, held wait and take-in-waits finding whose key the answer does not name, as a take-in.taken would. A seed that could not ask is not run, so ends nothing. Builds: design 47 "Taken in by the indexer it names" (A wait is bounded, said and recoverable: "At its start the watchman seeds its open waits from indexer_pending, and ends every take-in wait and take-in-waits finding it holds whose key indexer_pending no longer names"); 1966/AC195-5 Claude-Session: https://claude.ai/code/session_01JG73KEq4zoDfM9WtYpcQgZ
This commit is contained in:
@@ -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))
|
||||
|
||||
@@ -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})
|
||||
|
||||
Reference in New Issue
Block a user