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) + } + }) + } +}