From c7bddbbeabde1f8e1e6a7645541d27ce17a50eeb Mon Sep 17 00:00:00 2001 From: jochen Date: Fri, 23 Oct 2026 16:58:26 +0200 Subject: [PATCH] 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) --- .../internal/watch/mirror.go | 2 +- .../internal/watch/watch.go | 28 +++++++++++++-- .../internal/watch/watch_test.go | 36 ++++++++++++++++++- 3 files changed, 61 insertions(+), 5 deletions(-) diff --git a/modules/mesh-delivery-watchman/internal/watch/mirror.go b/modules/mesh-delivery-watchman/internal/watch/mirror.go index 485162c..6711785 100644 --- a/modules/mesh-delivery-watchman/internal/watch/mirror.go +++ b/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) } diff --git a/modules/mesh-delivery-watchman/internal/watch/watch.go b/modules/mesh-delivery-watchman/internal/watch/watch.go index f62d417..a5c8bee 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch.go +++ b/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 diff --git a/modules/mesh-delivery-watchman/internal/watch/watch_test.go b/modules/mesh-delivery-watchman/internal/watch/watch_test.go index 8abd76b..eef647d 100644 --- a/modules/mesh-delivery-watchman/internal/watch/watch_test.go +++ b/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) + } +}