diff --git a/modules/mesh-delivery-builder/buildrun/asks.go b/modules/mesh-delivery-builder/buildrun/asks.go index eb0b219f..bb60a103 100644 --- a/modules/mesh-delivery-builder/buildrun/asks.go +++ b/modules/mesh-delivery-builder/buildrun/asks.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "strings" "sync" "time" @@ -79,6 +80,7 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e } }() running, since, phase, phaseAt := "", time.Time{}, "", time.Time{} + var agains []string // the node of each run that ended on a transport fault, in order (ADR 0664 §3) 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) @@ -102,7 +104,24 @@ func (k Asker) Wait(ctx context.Context, a buildstep.StepAsk) (buildstep.Step, e if seen := stepSeenOf(ctx); seen != nil { seen(step) } - if step.State != buildstep.StepRunning { + if step.State == buildstep.StepAgain { + // A run ended on a transport fault and the bus gives the ask again: not an end. The bound runs on from the + // time it is given again, so a bus that never gives it says so (ADR 0664 §3). + agains = append(agains, step.Node) + running, since, phase, phaseAt = step.Node, step.Started, "given again after a transport fault", step.AgainAt + if k.Bound > 0 { + if timer != nil { + timer.Stop() + } + timer = time.NewTimer(max(time.Until(step.AgainAt.Add(k.Bound)), 0)) + bound = timer.C + } + continue + } + if buildstep.Ended(step) { + if step.Passing && step.State == buildstep.StepFailed && len(agains) > 0 { + step.Reason = fmt.Sprintf("a transport fault on %d runs (%s): %s", len(agains)+1, strings.Join(append(agains, step.Node), ", "), step.Reason) + } return step, nil } if k.Bound <= 0 { diff --git a/modules/mesh-delivery-builder/buildrun/asks_test.go b/modules/mesh-delivery-builder/buildrun/asks_test.go index 50c3e4cf..71694f56 100644 --- a/modules/mesh-delivery-builder/buildrun/asks_test.go +++ b/modules/mesh-delivery-builder/buildrun/asks_test.go @@ -40,6 +40,37 @@ func TestAnAskerPutsOnceAndReadsTheStep(t *testing.T) { if _, err := k.Ask(context.Background(), goAsk("mvl0frv18d1958")); err == nil || !strings.Contains(err.Error(), "shanks refused: go 1.27 installed") { t.Fatalf("a refused step: %v", err) } + // Issue 1748 (ADR 0664 §3): a run that ended on a transport fault is no end; the next run's end is the answer, in either + // arrival order; three runs then failed say a transport fault on 3 runs with each node; a bus that never gives the ask + // again ends the wait saying so. + for name, order := range map[string][]buildstep.Step{ + "again then passed": {{ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}, {ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}}, + "passed then again": {{ID: "mvl0frv18d1971", Node: "g14", State: buildstep.StepPassed, Run: 2}, {ID: "mvl0frv18d1971", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now()}}, + } { + h := &FakeAsks{} + for _, st := range order { + h.Write(st) + } + if got, err := (Asker{Publisher: h, Waiter: h, Bound: time.Minute}).Wait(context.Background(), goAsk("mvl0frv18d1971")); err != nil || got.State != buildstep.StepPassed { + t.Errorf("%s: %+v, %v", name, got, err) + } + } + h := &FakeAsks{} + for i, node := range []string{"ace", "g14", "shanks"} { + st := buildstep.Step{ID: "mvl0frv18d1972", Node: node, State: buildstep.StepAgain, Run: i + 1, AgainAt: time.Now()} + if i == 2 { + st.State, st.Passing, st.Reason = buildstep.StepFailed, true, "lookup proxy: i/o timeout" + } + h.Write(st) + } + if _, err := (Asker{Publisher: h, Waiter: h, Bound: time.Minute}).Ask(context.Background(), goAsk("mvl0frv18d1972")); err == nil || !strings.Contains(err.Error(), "a transport fault on 3 runs (ace, g14, shanks): lookup proxy: i/o timeout") { + t.Errorf("three runs then failed: %v", err) + } + h = &FakeAsks{} + h.Write(buildstep.Step{ID: "mvl0frv18d1973", Node: "ace", State: buildstep.StepAgain, Run: 1, AgainAt: time.Now(), Started: time.Now()}) + if _, err := (Asker{Publisher: h, Waiter: h, Bound: 20 * time.Millisecond}).Wait(context.Background(), goAsk("mvl0frv18d1973")); err == nil || !strings.Contains(err.Error(), "given again after a transport fault") { + t.Errorf("an ask never given again: %v", err) + } // review round 2 item 1: an ask that waits in its stream for a free slot is not bounded; once a node writes its step // running, the step is bounded from its start, and the error names the node. g := &FakeAsks{} diff --git a/modules/mesh-delivery-builder/buildrun/heardends.go b/modules/mesh-delivery-builder/buildrun/heardends.go index 3cae74f1..bdca8068 100644 --- a/modules/mesh-delivery-builder/buildrun/heardends.go +++ b/modules/mesh-delivery-builder/buildrun/heardends.go @@ -112,7 +112,7 @@ func (h *HeardEnds) WatchAsk(ctx context.Context, module, askID string) (<-chan continue } for _, s := range steps { - if s.State != buildstep.StepRunning { + if buildstep.Ended(s) { h.logf("the ask %s's step on %s ended %s, read at a nudge: its watch had not handed it over", askID, s.Node, s.State) } if !send(s) { @@ -206,7 +206,7 @@ func (h *HeardEnds) follow(ctx context.Context, kv jetstream.KeyValue, module st continue } heard = true - if step, err := buildstep.DecodeStep(e.Value()); err == nil && step.State != buildstep.StepRunning { + if step, err := buildstep.DecodeStep(e.Value()); err == nil && buildstep.Ended(step) { h.Nudge(module) } } diff --git a/modules/mesh-delivery-checker/internal/check/read.go b/modules/mesh-delivery-checker/internal/check/read.go index 4b0b8536..54ee1798 100644 --- a/modules/mesh-delivery-checker/internal/check/read.go +++ b/modules/mesh-delivery-checker/internal/check/read.go @@ -710,7 +710,7 @@ func (s stepsRead) lostWhy(c meshcatalog.CandidatesByGroupValue) string { var last *buildstep.Step for i := range s.all { st := s.all[i].step - if st.State != buildstep.StepRunning && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) { + if buildstep.Ended(st) && asksOf(c, st.What) && (last == nil || st.Ended.After(last.Ended)) { last = &s.all[i].step } } diff --git a/modules/mesh-delivery-checker/internal/check/try.go b/modules/mesh-delivery-checker/internal/check/try.go index 862a7a39..949a27a7 100644 --- a/modules/mesh-delivery-checker/internal/check/try.go +++ b/modules/mesh-delivery-checker/internal/check/try.go @@ -231,7 +231,7 @@ func (t Try) HoldsANode() bool { if len(t.Steps) == 0 { return true } - return slices.ContainsFunc(t.Steps, func(s TryStep) bool { return s.State == buildstep.StepRunning }) + return slices.ContainsFunc(t.Steps, func(s TryStep) bool { return s.State == buildstep.StepRunning || s.State == buildstep.StepAgain }) } // Over says whether the try came to its end. diff --git a/modules/mesh-delivery-checker/internal/check/try_test.go b/modules/mesh-delivery-checker/internal/check/try_test.go index b208cae6..a134e869 100644 --- a/modules/mesh-delivery-checker/internal/check/try_test.go +++ b/modules/mesh-delivery-checker/internal/check/try_test.go @@ -128,6 +128,10 @@ func TestATryIsAskedQueuedOncePerHeadAndAtMostFourAtOnce(t *testing.T) { // ADR t.Errorf("%s: answered %v, want refused %q", c.what, err, c.refused) } } + // Issue 1748: a try whose step ended on a transport fault, the bus giving its ask again, still runs. + if !(Try{State: TryRunning, Steps: []TryStep{{Ask: "a", State: buildstep.StepAgain}}}).HoldsANode() { + t.Error("a try with a step given again no longer runs") + } } func TestATryOfARepositoryWithNoCheckedLevelIsRefused(t *testing.T) { // the agreed error of the try diff --git a/sdk/buildrun/stale_test.go b/sdk/buildrun/stale_test.go index e9f570b8..9ceea9af 100644 --- a/sdk/buildrun/stale_test.go +++ b/sdk/buildrun/stale_test.go @@ -83,6 +83,7 @@ func TestEndStaleAndEndOne(t *testing.T) { put("ace", "raced1", "build_go_test", buildstep.StepRunning, boot.Add(-time.Hour), "") put("ace", "new1", "build_go_fetch", buildstep.StepRunning, boot.Add(time.Second), "") put("ace", "done1", "build_go_fetch", buildstep.StepPassed, boot.Add(-time.Hour), gone) + put("ace", "again1", "build_go_fetch", buildstep.StepAgain, boot.Add(-time.Hour), gone) // issue 1748: the bus gives it again; the sweep leaves it put("g14", "other1", "build_go_fetch", buildstep.StepRunning, boot.Add(-time.Hour), gone) // raced1 is ended by its own run between the read and the write. raced := false @@ -97,6 +98,9 @@ func TestEndStaleAndEndOne(t *testing.T) { if err != nil || len(ended) != 4 { t.Fatalf("ended %v, %v", ended, err) } + if s := k.last("ace.again1"); s.State != buildstep.StepAgain { + t.Errorf("the sweep ended a step the bus gives again: %+v", s) + } if _, held := k.puts["ace.mjk3l9x0abcded"]; held { t.Errorf("the ghost ask's key is still held: %+v", k.last("ace.mjk3l9x0abcded")) }