Bound a running build step's pulse by the step's own bound from its start, since nothing rewrites the key while it runs (issue 1908)

modules/mesh-delivery-watchman/internal/watch/mirror.go

@@ -50,5 +50,5 @@ func (w *Watchman) Step(language, key string, value []byte, op Op) {
 	if since.IsZero() {
 		since = w.now()
 	}
-	w.mirror(bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0)
+	w.mirrorWithin(w.stepBound, bucket, id, describe("the ask %s to end on %s", id, node), node, since, 0)
 }

modules/mesh-delivery-watchman/internal/watch/watch.go

@@ -58,12 +58,14 @@ type Watchman struct {
 	pulses   map[string]*pulse // live pulses, by pulse key
 	stuck    map[string]delivery.Stuck
 	bound    time.Duration
+	// stepBound is how long a build step may run once a node took the ask; a test shortens it.
+	stepBound time.Duration
 }
 
 // New is a watchman writing to out, reading the time from now.
 func New(out Out, now func() time.Time) *Watchman {
 	return &Watchman{out: out, now: now, counts: map[string]int{}, presence: map[string]presence{},
-		pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, bound: delivery.PulseMoveBound}
+		pulses: map[string]*pulse{}, stuck: map[string]delivery.Stuck{}, bound: delivery.PulseMoveBound, stepBound: StepBound}
 }
 
 // Logf sets where a write that failed is said; it is silent until set.
@@ -78,6 +80,11 @@ func (w *Watchman) say(format string, args ...any) {
 // mirror makes the pulse of the move (bucket, key) that began at since, unless it stands already for the same move.
 // A move that is already as old as the bound is stuck at once and has no pulse.
 func (w *Watchman) mirror(bucket, key, waited, head string, since time.Time, revision uint64) {
+	w.mirrorWithin(0, bucket, key, waited, head, since, revision)
+}
+
+// mirrorWithin is mirror for a move that may take bound (zero: the watchman's own bound, delivery.PulseMoveBound).
+func (w *Watchman) mirrorWithin(bound time.Duration, bucket, key, waited, head string, since time.Time, revision uint64) {
 	pk, err := delivery.PulseKey(bucket, key)
 	if err != nil {
 		w.say("no pulse for %s %s: %v", bucket, key, err)
@@ -89,6 +96,9 @@ func (w *Watchman) mirror(bucket, key, waited, head string, since time.Time, rev
 		return
 	}
 	ttl, stalled := w.ttl(w.now().Sub(since))
+	if bound > 0 {
+		ttl, stalled = bound-w.now().Sub(since), w.now().Sub(since) >= bound
+	}
 	if stalled {
 		if tracked {
 			w.deletePulse(pk)
@@ -219,11 +229,23 @@ func (w *Watchman) Reconcile() {
 
 func describe(format string, args ...any) string { return fmt.Sprintf(format, args...) }
 
-// SetBound sets how long a move may take; it is delivery.PulseMoveBound unless a test shortens it.
+// StepBound is how long a build step may run, from its start on the node, before it is stuck: the asker's bound on a step
+// (mesh-sdk go/delivery/buildrun Asks.Bound, 30 minutes for a build). Nothing rewrites a running step's key, so its pulse
+// is bounded by this and not by the 120 s of a move that is heartbeat (issue 1908).
+const StepBound = 30 * time.Minute
+
+// SetStepBound sets how long a build step may run; it is StepBound unless a test shortens it.
+func (w *Watchman) SetStepBound(d time.Duration) {
+	w.mu.Lock()
+	defer w.mu.Unlock()
+	w.stepBound = d
+}
+
+// SetBound sets how long a move, and a running step, may take; it is delivery.PulseMoveBound unless a test shortens it.
 func (w *Watchman) SetBound(d time.Duration) {
 	w.mu.Lock()
 	defer w.mu.Unlock()
-	w.bound = d
+	w.bound, w.stepBound = d, d
 }
 
 // ttl is the time to live of a pulse for a move of the given age, and whether the move is stalled already

modules/mesh-delivery-watchman/internal/watch/watch_test.go

@@ -63,7 +63,9 @@ func (c *clock) now() time.Time { return c.t }
 
 func setup() (*Watchman, *fakeOut, *clock) {
 	o, c := newOut(), &clock{t: t0}
-	return New(o, c.now), o, c
+	w := New(o, c.now)
+	w.SetStepBound(delivery.PulseMoveBound) // the older tests move a step as they move any key: stuck at the pulse's bound
+	return w, o, c
 }
 
 // checkValue is a check's key as the controller's builder writes it (mesh-controller internal/link/checks_nats.go
@@ -497,3 +499,35 @@ func TestAPausedKeyOrAnotherNodesNonStepIsLeft(t *testing.T) { // 1908
 		t.Errorf("pulses %v stuck %v", o.pulses, o.stuck)
 	}
 }
+
+func TestARunningStepOfTenMinutesUnderItsBoundIsNotStuck(t *testing.T) { // 1908
+	o, c := newOut(), &clock{t: t0}
+	w := New(o, c.now) // the step's own bound: 30 minutes
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
+	c.t = t0.Add(10 * time.Minute)
+	// no heartbeat is written, and the server expires the pulse at its ttl: 30 minutes from the start, not 120 s
+	if len(o.stuck) != 0 {
+		t.Fatalf("a build of 10 minutes is stuck: %v", o.stuck)
+	}
+	if got := o.ttls[pulseKeyOf(t, checksKey, "askid1")]; got != 30*time.Minute {
+		t.Errorf("ttl = %v, want the step's bound of 30m from its start", got)
+	}
+}
+
+func TestARunningStepPastItsBoundIsStuck(t *testing.T) { // 1908
+	o, c := newOut(), &clock{t: t0}
+	w := New(o, c.now)
+	w.Step("go", "n1.askid1", runningStep("askid1", t0), Put)
+	c.t = t0.Add(31 * time.Minute)
+	w.Pulse(pulseKeyOf(t, checksKey, "askid1"), Expired, c.t)
+	s, ok := o.stuck[stuckKeyOf(t, checksKey, "askid1")]
+	if !ok || s.Moved || !s.Since.Equal(t0) {
+		t.Fatalf("stuck = %+v, %v", s, ok)
+	}
+	// and one found past its bound at the watchman's start is stuck at once
+	o2, c2 := newOut(), &clock{t: t0.Add(time.Hour)}
+	New(o2, c2.now).Step("go", "n1.askid1", runningStep("askid1", t0), Put)
+	if _, ok := o2.stuck[stuckKeyOf(t, checksKey, "askid1")]; !ok || len(o2.pulses) != 0 {
+		t.Errorf("stuck %v pulses %v", o2.stuck, o2.pulses)
+	}
+}