diff --git a/internal/planning/executor.go b/internal/planning/executor.go index eaf7736..4a3ad46 100644 --- a/internal/planning/executor.go +++ b/internal/planning/executor.go @@ -800,8 +800,8 @@ func (e *Executor) reconsider(ctx context.Context, p Plan, it volition.Intention rc := volition.Reconsider(it, t, volition.ReconsiderContext{Replans: p.Revision - 1, MaxReplans: p.MaxReplans, Now: e.now()}) switch rc.Outcome { case volition.Replan, volition.Tighten: - if err := e.Plans.SetStatus(ctx, nil, p.ID, PlanReplanning, rc.Reason); err != nil { - return err + if won, err := e.Plans.BeginReplan(ctx, p.ID, rc.Reason); err != nil || !won { + return err // another reconsideration is already replanning or has closed this plan } cur, err := e.Volition.Apply(ctx, r, p.MindID, it.ID, rc, t) if err != nil { diff --git a/internal/planning/executor_test.go b/internal/planning/executor_test.go index a51e0ce..12861d0 100644 --- a/internal/planning/executor_test.go +++ b/internal/planning/executor_test.go @@ -538,3 +538,22 @@ func TestStaleStepCopyCannotMoveStepBackwards(t *testing.T) { t.Fatalf("stale copy overwrote a claimed step: %s", got.Status) } } + +// Two reconsiderations holding the same plan (an event reaction and an +// executor tick) must replan it once, not twice. +func TestConcurrentReconsiderationReplansOnce(t *testing.T) { + f := newFixture(t) + f.allow("service.restart") + it := f.intention(false) + p := f.start(it) + trig := volition.Trigger{Kind: volition.TriggerStepFailed, Detail: "test"} + if err := f.exec.reconsider(f.ctx, p, it, trig); err != nil { + t.Fatal(err) + } + if err := f.exec.reconsider(f.ctx, p, it, trig); err != nil { // stale copies of the plan and intention + t.Fatal(err) + } + if ps, _ := f.plans.ForIntention(f.ctx, nil, it.ID); len(ps) != 2 { + t.Fatalf("want one replan, got %d plans", len(ps)) + } +} diff --git a/internal/planning/store.go b/internal/planning/store.go index 25dc130..a04329e 100644 --- a/internal/planning/store.go +++ b/internal/planning/store.go @@ -399,6 +399,18 @@ func (s *Store) SetStatus(ctx context.Context, q storage.Querier, id ids.ID, sta return err } +// BeginReplan moves an open plan into replanning only if no one else has, +// so two reconsiderations holding the same plan cannot both replan it. It +// reports whether this caller won. +func (s *Store) BeginReplan(ctx context.Context, id ids.ID, reason string) (bool, error) { + tag, err := s.db.Exec(ctx, `UPDATE plans SET status=$2, status_reason=$3, updated_at=now() WHERE id=$1 AND status = ANY($4)`, + id, PlanReplanning, trunc(reason, 500), []string{PlanValidated, PlanWaiting, PlanExecuting, PlanBlocked, PlanAwaitingApproval}) + if err != nil { + return false, err + } + return tag.RowsAffected() == 1, nil +} + // AddActual adds consumed resources to a plan and returns the new totals. func (s *Store) AddActual(ctx context.Context, id ids.ID, delta Resources) (Resources, error) { var raw []byte