Restart an ask's bound at each run's start, also on the same node, so a step redelivered after its build module was replaced is not cut by the killed run's time (issue 1887, review: issue 734 path; ADR 0525 §6)
This commit is contained in:
@@ -75,7 +75,7 @@ func (k Asker) Wait(ctx context.Context, a delivery.StepAsk) (delivery.Step, err
|
||||
timer.Stop()
|
||||
}
|
||||
}()
|
||||
running := ""
|
||||
running, since := "", time.Time{}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
@@ -89,11 +89,12 @@ 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 {
|
||||
if k.Bound <= 0 || (step.Node == running && step.Started.Equal(since)) {
|
||||
continue
|
||||
}
|
||||
// A node took the ask (or another took it after the first stopped): its step is bounded from its start.
|
||||
running = step.Node
|
||||
// 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
|
||||
left := k.Bound
|
||||
if !step.Started.IsZero() {
|
||||
left = time.Until(step.Started.Add(k.Bound))
|
||||
|
||||
@@ -49,3 +49,23 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) {
|
||||
t.Fatal("a wait put an ask")
|
||||
}
|
||||
}
|
||||
|
||||
// Review (issue 734 path): an ask taken again by the same node (its build module replaced mid-run) is bounded from the
|
||||
// new run's start, so its pass is answered after the first run's bound would have ended the wait.
|
||||
func TestARedeliveryToTheSameNodeRestartsTheBound(t *testing.T) {
|
||||
f := &FakeAsks{}
|
||||
k := Asker{Publisher: f, Waiter: f, Bound: 60 * time.Millisecond}
|
||||
const id = "mvl0frv18d1962"
|
||||
first := time.Now()
|
||||
f.Write(delivery.Step{ID: id, Node: "ace", State: delivery.StepRunning, Started: first})
|
||||
go func() {
|
||||
time.Sleep(40 * 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()})
|
||||
}()
|
||||
step, err := k.Wait(context.Background(), goAsk(id))
|
||||
if err != nil || step.State != delivery.StepPassed {
|
||||
t.Fatalf("step %+v, %v, want the redelivered run's pass", step, err)
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user