From 672884cd229537ecffd75dfc825252a0733b9ac3 Mon Sep 17 00:00:00 2001 From: Andy Dixon Date: Sun, 20 Sep 2026 22:54:24 +0100 Subject: [PATCH] fix(planning): a stale step copy could reset a claimed step and run it twice MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The pending → ready and pending → cancelled transitions, and cancellation of open steps, saved a step unconditionally. Under load an executor holding a stale "pending" copy could write "ready" over a step another executor had already claimed and was running, after which the step could be claimed again: the full race suite caught two restarts from one plan. These transitions now use SaveIf, which writes only when the stored status is still the expected one. A deterministic regression test hands the executor a stale copy and fails without the fix; the concurrency test now races eight executors ticking repeatedly. --- internal/planning/executor.go | 7 +++--- internal/planning/executor_test.go | 39 +++++++++++++++++++++++++++--- internal/planning/store.go | 18 ++++++++++++++ 3 files changed, 58 insertions(+), 6 deletions(-) diff --git a/internal/planning/executor.go b/internal/planning/executor.go index 97ca5ec..eaf7736 100644 --- a/internal/planning/executor.go +++ b/internal/planning/executor.go @@ -247,14 +247,14 @@ func (e *Executor) advanceStep(ctx context.Context, p *Plan, it volition.Intenti } if dep.Status == StepFailed || dep.Status == StepCancelled { st.Status, st.StatusReason = StepCancelled, "dependency "+d+" did not complete" - return true, e.Plans.Save(ctx, nil, *st) + return e.Plans.SaveIf(ctx, *st, StepPending) } if !StepDone(dep.Status) { return false, nil } } st.Status = StepReady - return true, e.Plans.Save(ctx, nil, *st) + return e.Plans.SaveIf(ctx, *st, StepPending) case StepReady: if !due || it.Status != volition.IntentionActive { return false, nil @@ -857,8 +857,9 @@ func (e *Executor) cancelOpenSteps(ctx context.Context, p Plan, reason string) { e.Caps.WithdrawCapabilityRequest(ctx, p.MindID, s.RequestID, reason) //nolint:errcheck } e.dropSubordinate(ctx, &p, &s, reason) + from := s.Status s.Status, s.StatusReason = StepCancelled, reason - e.Plans.Save(ctx, nil, s) //nolint:errcheck + e.Plans.SaveIf(ctx, s, from) //nolint:errcheck // a step that moved on meanwhile keeps its newer state } } diff --git a/internal/planning/executor_test.go b/internal/planning/executor_test.go index 14d1708..a51e0ce 100644 --- a/internal/planning/executor_test.go +++ b/internal/planning/executor_test.go @@ -415,13 +415,17 @@ func TestConcurrentExecutorsRunAStepOnce(t *testing.T) { f.host.fixOnRestart = true it := f.intention(false) p := f.start(it) - other := f.newExecutor(f.exec.Predictions, f.exec.Planner, f.exec.Log) + // Many executors race over the same plan from its first transition, each + // ticking repeatedly, as duplicate schedulers or several daemons would. var wg sync.WaitGroup - for _, e := range []*Executor{f.exec, other, f.newExecutor(f.exec.Predictions, f.exec.Planner, f.exec.Log)} { + for i := 0; i < 8; i++ { + e := f.newExecutor(f.exec.Predictions, f.exec.Planner, f.exec.Log) wg.Add(1) go func(e *Executor) { defer wg.Done() - e.TickMind(f.ctx, f.mind.ID) //nolint:errcheck + for j := 0; j < 3; j++ { + e.TickMind(f.ctx, f.mind.ID) //nolint:errcheck + } }(e) } wg.Wait() @@ -505,3 +509,32 @@ func TestInvalidModelPlanNeverRuns(t *testing.T) { t.Fatal("invalid plan executed") } } + +// Regression: an executor holding a stale "pending" copy of a step must not +// move it back to "ready" after another executor has claimed it, or the +// step can be claimed and run a second time. +func TestStaleStepCopyCannotMoveStepBackwards(t *testing.T) { + f := newFixture(t) + f.allow("service.restart") + it := f.intention(false) + p := f.start(it) + stale := f.step(p, "P1") // pending + fresh := stale + fresh.Status = StepReady + if ok, err := f.plans.SaveIf(f.ctx, fresh, StepPending); err != nil || !ok { + t.Fatalf("pending → ready: %v %v", ok, err) + } + if ok, err := f.plans.Claim(f.ctx, &fresh, StepReady, StepExecuting, time.Minute); err != nil || !ok { + t.Fatalf("claim: %v %v", ok, err) + } + byKey := map[string]*Step{} + for i := range p.Steps { + byKey[p.Steps[i].Key] = &p.Steps[i] + } + if _, err := f.exec.advanceStep(f.ctx, &p, it, &stale, byKey); err != nil { + t.Fatal(err) + } + if got := f.step(p, "P1"); got.Status != StepExecuting { + t.Fatalf("stale copy overwrote a claimed step: %s", got.Status) + } +} diff --git a/internal/planning/store.go b/internal/planning/store.go index 2cda720..25dc130 100644 --- a/internal/planning/store.go +++ b/internal/planning/store.go @@ -367,6 +367,24 @@ func (s *Store) Save(ctx context.Context, q storage.Querier, st Step) error { return err } +// SaveIf writes a step only if its stored status is still expected, so a +// caller holding a stale copy can never move a step backwards (for example +// resetting a step another executor is running to "ready"). It reports +// whether the write happened. +func (s *Store) SaveIf(ctx context.Context, st Step, expected string) (bool, error) { + now := time.Now() + var finished *time.Time + if StepTerminal(st.Status) { + finished = &now + } + tag, err := s.db.Exec(ctx, `UPDATE plan_steps SET status=$2, status_reason=$3, finished_at=coalesce($4, finished_at), updated_at=$5 WHERE id=$1 AND status=$6`, + st.ID, st.Status, trunc(st.StatusReason, 500), finished, now, expected) + if err != nil { + return false, err + } + return tag.RowsAffected() == 1, nil +} + // SetStatus writes a plan's status and reason. func (s *Store) SetStatus(ctx context.Context, q storage.Querier, id ids.ID, status, reason string) error { if q == nil {