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.
2026-10-30 18:25:56 +01:00
parent ff58b16cf9
commit 255f8b2aca
4 changed files with 96 additions and 2 deletions
@@ -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 `<node>.<ask id>` 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.
@@ -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)
@@ -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.
@@ -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)
+			}
+		})
+	}
+}