From 255f8b2aca1190b5955f068f8834c7a4c0d45c70 Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 30 Oct 2026 18:25:56 +0100 Subject: [PATCH] Close a stuck key whose watched key aged out or was deleted unheard, so its condition clears (issue 774) A key that ages out of its bucket (the build seats' asks pass the stream's 168 h) leaves no marker a watch hears, and one deleted while the watchman was down is not heard either: its stuck key stood unmoved for ever, and 26 stuck-key conditions of asks from 2026-10-21 were open on 10-30. Quick fix: mark stuck keys older than the stream's maximum age moved. Thorough fix, chosen because it covers every bucket the watchman mirrors and any way a key goes: at the start's end, a stuck key not moved whose bucket was read whole at this start (each build seat's steps, the assigner's wants) and whose key was not read standing is written moved, which clears its finding, and the closing is said in the log. A bucket that could not be read closes nothing. A key that ages out while the watchman runs is closed at its next start: a KV age-out emits no event, and no timer is added. --- .../cmd/mesh-delivery-watchman/main.go | 5 ++ .../internal/watch/mirror.go | 2 + .../internal/watch/watch.go | 51 ++++++++++++++++++- .../internal/watch/watch_test.go | 40 +++++++++++++++ 4 files changed, 96 insertions(+), 2 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 660ea74..95af732 100644 --- a/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go +++ b/modules/mesh-delivery-watchman/cmd/mesh-delivery-watchman/main.go @@ -244,6 +244,8 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro process.NotWatched(logger.Printf, delivery.AssignerWantsBucket, err, "moves are not watched until the watchman starts again") haveSeen.Stop() haveSeen = nil + } else { + w.ReadWhole(delivery.AssignerWantsBucket) } } // The build seats' steps: a step key `.` written is the ask taken (issue 1887 A8b). A seat whose build module @@ -263,6 +265,9 @@ func run(ctx context.Context, getenv func(string) string, logOut io.Writer) erro continue } taken[lang] = sw + if bucket, err := delivery.AskStream(lang); err == nil { + w.ReadWhole(bucket) + } } // The events first, then the one request of the open connections, so no connection falls between them. diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index fe9f262..5bf05e7 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/modules/mesh-delivery-watchman/internal/watch/mirror.go @@ -40,6 +40,7 @@ func (w *Watchman) Step(language, key string, value []byte, op Op) { w.moved(bucket, id) return } + w.stands(bucket, id) step, err := delivery.DecodeStep(value) if err != nil { w.say("the step %s cannot be read, and is left: %v", key, err) @@ -80,6 +81,7 @@ func (w *Watchman) Want(key string, value []byte, op Op, at time.Time) { w.moved(delivery.AssignerWantsBucket, key) return } + w.stands(delivery.AssignerWantsBucket, key) want, err := delivery.DecodeWant(value) if err != nil { w.say("the want %s cannot be read, and is left: %v", key, err) diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index cc4dcc3..fa6d5b7 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch.go @@ -91,6 +91,11 @@ type Watchman struct { tunnels map[string]string // stepBound is how long a build step may run once a node took the ask; a test shortens it. stepBound time.Duration + // standing is, until the start ends (Reconcile), each mirrored key read standing, by the bucket a stuck key names it + // under; readWhole the buckets whose keys were all read at this start (ReadWhole). reconciled is the start's end. + standing map[string]map[string]bool + readWhole map[string]bool + reconciled bool } // New is a watchman writing to out, reading the time from now. @@ -100,7 +105,8 @@ func New(out Out, now func() time.Time) *Watchman { findings: map[string][]byte{}, seededFindings: map[string]bool{}, signals: map[string]signalDue{}, 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{}} + health: map[string]delivery.NodeHealth{}, tunnels: map[string]string{}, + standing: map[string]map[string]bool{}, readWhole: map[string]bool{}} } // Logf sets where a write that failed is said; it is silent until set. @@ -257,7 +263,8 @@ func (w *Watchman) SeedPulse(key string) { } // Reconcile ends the start: a pulse that stood and that no mirrored key has made again is a move that happened while -// the watchman was away, so it is deleted and writes no stuck key. +// the watchman was away, so it is deleted and writes no stuck key; and a stuck key whose watched key is gone is closed +// (closeGone). func (w *Watchman) Reconcile() { w.mu.Lock() defer w.mu.Unlock() @@ -266,9 +273,49 @@ func (w *Watchman) Reconcile() { w.deletePulse(pk) } } + w.closeGone() w.reconcileFindings() } +// **A stuck key whose watched key is gone is closed** (issue 774, sweep 2026-10-30): a key that ages out of its bucket +// (a stream's maximum age, 168 h for the build seats' asks) leaves no marker a watch hears, and one deleted while the +// watchman was down is not heard either, so its stuck key stood unmoved for ever: 26 stuck-key conditions of asks from +// 2026-10-21 were open on 10-30. At the start's end, every stuck key not moved whose bucket was read whole at this start +// (ReadWhole) and whose key was not read standing is written moved, which clears its finding, and said. A bucket that +// could not be read closes nothing: its keys are not known to be gone. + +// ReadWhole says every key of bucket (as a stuck key names it: an ask stream for a build seat's steps, the assigner's +// wants) was read at this start; called once its watch holds what stands. +func (w *Watchman) ReadWhole(bucket string) { + w.mu.Lock() + defer w.mu.Unlock() + w.readWhole[bucket] = true +} + +// stands records a mirrored key read standing during the start. +func (w *Watchman) stands(bucket, key string) { + if w.reconciled { + return + } + if w.standing[bucket] == nil { + w.standing[bucket] = map[string]bool{} + } + w.standing[bucket][key] = true +} + +func (w *Watchman) closeGone() { + for sk, s := range w.stuck { + if s.Moved || !w.readWhole[s.Bucket] || w.standing[s.Bucket][s.Key] { + continue + } + s.Moved = true + w.say("the stuck key %s is closed: its key %s is no longer in %s (it aged out of the bucket or was deleted while nobody heard), so nothing waits for it since %s", + sk, s.Key, s.Bucket, s.Since.UTC().Format(time.RFC3339)) + w.putStuck(sk, s) + } + w.standing, w.readWhole, w.reconciled = map[string]map[string]bool{}, map[string]bool{}, true +} + // reconcileFindings ends the start for the findings: one read at the start that what the watchman read since did not // observe again is deleted, unless a signal still standing keeps it (a provider's failing word, an advisory's quiet hour, a // machine's silent tools, a silent self-check): those end by their own events or expiry. diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index edcdec9..ed96fa6 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/modules/mesh-delivery-watchman/internal/watch/watch_test.go @@ -656,3 +656,43 @@ func TestAWantNotAppliedByItsBoundIsStuckAndAKeyDeletedEndsIt(t *testing.T) { t.Errorf("a deleted want left its pulse: %v", o2.pulses) } } + +// Sweep 2026-10-30: a watched key that aged out of its bucket or was deleted while nobody heard leaves its stuck key; at the +// start's end the watchman closes it (moved, its finding cleared), but only for a bucket it read whole at that start. +func TestAStuckKeyWhoseKeyIsGoneIsClosedAtTheStart(t *testing.T) { + wantKey := "want.n1.mod-a" + for _, row := range []struct { + name string + bucket string + key string + standing func(w *Watchman) // what the start reads standing + readWhole bool + closed bool + }{ + {"an ask's step aged out of the steps bucket", checksKey, "askold1", func(*Watchman) {}, true, true}, + {"an ask's step still running", checksKey, "askold1", func(w *Watchman) { w.Step("go", "n1.askold1", runningStep("askold1", t0), Put) }, true, false}, + {"the steps bucket could not be read", checksKey, "askold1", func(*Watchman) {}, false, false}, + {"a want deleted while nobody heard", delivery.AssignerWantsBucket, wantKey, func(*Watchman) {}, true, true}, + {"a want still standing", delivery.AssignerWantsBucket, wantKey, func(w *Watchman) { w.Want(wantKey, wantValue(t, fpNew, false), Put, t0) }, true, false}, + } { + t.Run(row.name, func(t *testing.T) { + w, o, c := setup() + c.t = t0.Add(9 * 24 * time.Hour) + sk := stuckKeyOf(t, row.bucket, row.key) + v, _ := delivery.EncodeStuck(delivery.Stuck{Bucket: row.bucket, Key: row.key, Waited: "a move", Since: t0}) + w.SeedStuck(sk, v) + if _, ok := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)]; !ok { + t.Fatal("the seeded stuck key raised no finding") + } + row.standing(w) + if row.readWhole { + w.ReadWhole(row.bucket) + } + w.Reconcile() + _, open := o.findings[FindingKey(ScopeDelivery, sk, KindStuck)] + if got := o.stuck[sk].Moved; got != row.closed || open == row.closed { + t.Errorf("moved %v, finding open %v; want closed %v", got, open, row.closed) + } + }) + } +}