novox/mesh-delivery
19
0
Fork 0
Code Issues Pull Requests 2 Actions Packages Projects Releases Wiki Activity

Watchman: a replayed pending that its taken ends is never raised; a drop then a pending again is covered (issue 1585, ADR 0633 §5)

Browse Source 
Claude-Session: https://claude.ai/code/session_01JG73KG7AfSdh6FP8VrBqCn
This commit is contained in:
jschoubben
2026-11-04 23:09:20 +01:00
parent d11a5af113
commit 7609f2d9a2
2 changed files with 32 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files
modules/mesh-delivery-watchman/internal/watch/findings_test.go
+26 -1
View File
@@ -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),
modules/mesh-delivery-watchman/internal/watch/takein.go
+6 -3
View File
@@ -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})