novox/mesh-sdk
18
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

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)

Browse Source 
Claude-Session: https://claude.ai/code/session_01NcbgfJ9Gum47zkFKWRaovL
This commit is contained in:
jschoubben
2026-11-02 11:31:38 +01:00
parent 7b034e1db0
commit e240e01d07
2 changed files with 81 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
go/delivery/buildrun/asks.go
+45 -11
View File
@@ -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
}
}
}
}
go/delivery/buildrun/asks_test.go
+36 -1
View File
@@ -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()})