From e240e01d079a6798027c2fefc54fd03fd8da1e66 Mon Sep 17 00:00:00 2001 From: jochen Date: Mon, 2 Nov 2026 01:35:12 +0100 Subject: [PATCH] Bound an ask's step by how long it stands still, not its age, so a step moving through phases is waited for (issue 1930) Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL --- go/delivery/buildrun/asks.go | 56 +++++++++++++++++++++++++------ go/delivery/buildrun/asks_test.go | 37 +++++++++++++++++++- 2 files changed, 81 insertions(+), 12 deletions(-) diff --git a/go/delivery/buildrun/asks.go b/go/delivery/buildrun/asks.go index a03bbbf..fdaf305 100644 --- a/go/delivery/buildrun/asks.go +++ b/go/delivery/buildrun/asks.go @@ -32,8 +32,8 @@ type AskWaiter interface { type Asker struct { Publisher AskPublisher Waiter AskWaiter - // Bound is how long a step may run once a build node took the ask (its key written running) before the asker stops - // waiting; zero waits as long as ctx. An ask waiting in its stream is not bounded: the stream's backlog is no fault + // Bound is how long a running step (a build node took the ask, its key written running) may stand still, with no new phase, + // before the asker stops waiting; a step that keeps moving is waited for up to 3 Bounds from its start (issue 1930); zero waits as long as ctx. An ask waiting in its stream is not bounded: the stream's backlog is no fault // (ADR 0525 §6, §9), and the bound starts again when another node takes the ask. Bound time.Duration } @@ -68,20 +68,33 @@ func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, err if err != nil { return delivery.Step{}, err } - var bound <-chan time.Time - var timer *time.Timer + var bound, outerCap <-chan time.Time + var timer, capTimer *time.Timer defer func() { if timer != nil { timer.Stop() } + if capTimer != nil { + capTimer.Stop() + } }() - running, since := "", time.Time{} + running, since, phase, phaseAt := "", time.Time{}, "", time.Time{} + standStill := func() error { + if phase == "" { + return fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it stood still, saying no phase", a.ID, running, k.Bound) + } + return fmt.Errorf("the ask %s's step on %s did not end within %s: it stood still in %q since %s", + a.ID, running, k.Bound, phase, phaseAt.UTC().Format(time.RFC3339)) + } for { select { case <-ctx.Done(): return delivery.Step{}, ctx.Err() case <-bound: - return delivery.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start", a.ID, running, k.Bound) + return delivery.Step{}, standStill() + case <-outerCap: + return delivery.Step{}, fmt.Errorf("the ask %s's step on %s did not end within %s of its start: it kept moving (last phase %q at %s) but never ended", + a.ID, running, 3*k.Bound, phase, phaseAt.UTC().Format(time.RFC3339)) case step, ok := <-steps: if !ok { return delivery.Step{}, errors.New("the watch of the ask's step ended") @@ -92,21 +105,42 @@ func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, err if step.State != delivery.StepRunning { return step, nil } - if k.Bound <= 0 || (step.Node == running && step.Started.Equal(since)) { + if k.Bound <= 0 { + continue + } + newRun := step.Node != running || !step.Started.Equal(since) + if !newRun && step.Phase == phase && step.PhaseAt.Equal(phaseAt) { continue } // A node took the ask, or took it again after its run stopped (another node, or the same node's build module - // replaced mid-run: a new start): its step is bounded from its own start. - running, since = step.Node, step.Started + // replaced mid-run: a new start), or its step entered a phase. The bound is on standing still: it runs from the + // step's last movement (its phase's start, else its start), so a step that keeps moving is waited for, up to + // three bounds from its start (issue 1930). + running, since, phase, phaseAt = step.Node, step.Started, step.Phase, step.PhaseAt + last := step.Started + if step.PhaseAt.After(last) { + last = step.PhaseAt + } left := k.Bound - if !step.Started.IsZero() { - left = time.Until(step.Started.Add(k.Bound)) + if !last.IsZero() { + left = time.Until(last.Add(k.Bound)) } if timer != nil { timer.Stop() } timer = time.NewTimer(max(left, 0)) bound = timer.C + if newRun { + if capTimer != nil { + capTimer.Stop() + } + outer := 3 * k.Bound + if !step.Started.IsZero() { + outer = time.Until(step.Started.Add(3 * k.Bound)) + } + capTimer = time.NewTimer(max(outer, 0)) + outerCap = capTimer.C + } } } } diff --git a/go/delivery/buildrun/asks_test.go b/go/delivery/buildrun/asks_test.go index 7453b63..5403a54 100644 --- a/go/delivery/buildrun/asks_test.go +++ b/go/delivery/buildrun/asks_test.go @@ -53,6 +53,41 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if len(g.Asks()) != 0 { t.Fatal("a wait put an ask") } + // Issue 1930: the bound is on a step that stands still, not on its age. A step that enters a new phase inside the bound is + // waited for until it ends; one that stays in a phase the bound fails at once, naming the phase; one that never stops moving + // ends at three bounds, said as such. + k = Asker{Publisher: g, Waiter: g, Bound: 60 * time.Millisecond} + started := time.Now() + go func() { + for i := range 8 { + g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepRunning, Started: started, Phase: "p" + string(rune('a'+i)), PhaseAt: time.Now()}) + time.Sleep(15 * time.Millisecond) + } + g.Write(delivery.Step{ID: "mvl0frv18d1980", Node: "ace", State: delivery.StepPassed}) + }() + if step, err := k.Wait(context.Background(), goAsk("mvl0frv18d1980")); err != nil || step.State != delivery.StepPassed { + t.Fatalf("a step that kept moving past its bound: %+v, %v", step, err) + } + g.Write(delivery.Step{ID: "mvl0frv18d1981", Node: "ace", State: delivery.StepRunning, Started: time.Now(), Phase: "integration: tests", PhaseAt: time.Now()}) + if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1981")); err == nil || !strings.Contains(err.Error(), `stood still in "integration: tests"`) { + t.Fatalf("a step that stood still in a phase: %v", err) + } + loop := make(chan struct{}) + defer close(loop) + started = time.Now() + go func() { + for i := 0; ; i++ { + select { + case <-loop: + return + case <-time.After(20 * time.Millisecond): + g.Write(delivery.Step{ID: "mvl0frv18d1982", Node: "ace", State: delivery.StepRunning, Started: started, Phase: "q" + string(rune('a'+i%26)), PhaseAt: time.Now()}) + } + } + }() + if _, err := k.Wait(context.Background(), goAsk("mvl0frv18d1982")); err == nil || !strings.Contains(err.Error(), "kept moving") { + t.Fatalf("a step that never stops moving: %v", err) + } // Issue 734: an ask the build seat gave up on (DEAD_LETTERS keeps its copy with the ask's id) ends its wait as failed // with the words, though a node had taken it; under NotTakenBound one no node takes ends at the bound; a step written // first is answered as the node wrote it. @@ -157,7 +192,7 @@ func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) { first := time.Now() f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: first}) go func() { - time.Sleep(40 * time.Millisecond) + time.Sleep(15 * time.Millisecond) f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: time.Now()}) time.Sleep(50 * time.Millisecond) // past the first run's bound, inside the second's f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepPassed, Started: time.Now()})