diff --git a/internal/planning/executor_test.go b/internal/planning/executor_test.go index 95cadd6..d794bad 100644 --- a/internal/planning/executor_test.go +++ b/internal/planning/executor_test.go @@ -591,3 +591,28 @@ func TestConcurrentReconsiderationReplansOnce(t *testing.T) { t.Fatalf("want one replan, got %d plans", len(ps)) } } + +// Regression: a caller holding a stale copy of a plan (a reconsideration +// that lost a race with completion) must not relabel a closed plan. +func TestClosedPlanIsNeverRelabelled(t *testing.T) { + f := newFixture(t) + f.host.fixOnRestart = true + f.allow("service.restart") + it := f.intention(false) + p := f.start(it) + for i := 0; i < 5 && f.plan(p.ID).Status != PlanCompleted; i++ { + f.tick() + } + if f.plan(p.ID).Status != PlanCompleted { + t.Fatalf("plan did not complete: %s", f.plan(p.ID).Status) + } + if err := f.plans.SetStatus(f.ctx, nil, p.ID, PlanBlocked, "stale reconsideration"); err != nil { + t.Fatal(err) + } + if err := f.exec.Reconsider(f.ctx, p.ID, volition.Trigger{Kind: volition.TriggerResourceExhausted, Detail: "late"}); err != nil { + t.Fatal(err) + } + if got := f.plan(p.ID); got.Status != PlanCompleted || got.StatusReason == "stale reconsideration" { + t.Fatalf("closed plan relabelled: %+v", got) + } +} diff --git a/internal/planning/store.go b/internal/planning/store.go index a04329e..92dae04 100644 --- a/internal/planning/store.go +++ b/internal/planning/store.go @@ -385,7 +385,10 @@ func (s *Store) SaveIf(ctx context.Context, st Step, expected string) (bool, err return tag.RowsAffected() == 1, nil } -// SetStatus writes a plan's status and reason. +// SetStatus writes a plan's status and reason. Only an open plan (or a +// draft) changes: a plan that was completed, failed, abandoned or +// superseded meanwhile keeps that state, so a caller holding a stale copy +// can never reopen or relabel a closed plan. func (s *Store) SetStatus(ctx context.Context, q storage.Querier, id ids.ID, status, reason string) error { if q == nil { q = s.db @@ -395,7 +398,8 @@ func (s *Store) SetStatus(ctx context.Context, q storage.Querier, id ids.ID, sta if !PlanOpen(status) && status != PlanDraft { closed = &now } - _, err := q.Exec(ctx, `UPDATE plans SET status=$2, status_reason=$3, closed_at=coalesce($4, closed_at), updated_at=$5 WHERE id=$1`, id, status, trunc(reason, 500), closed, now) + _, err := q.Exec(ctx, `UPDATE plans SET status=$2, status_reason=$3, closed_at=coalesce($4, closed_at), updated_at=$5 WHERE id=$1 AND status = ANY($6)`, + id, status, trunc(reason, 500), closed, now, []string{PlanDraft, PlanValidated, PlanWaiting, PlanExecuting, PlanBlocked, PlanAwaitingApproval, PlanReplanning}) return err } diff --git a/internal/volition/store.go b/internal/volition/store.go index 738db1f..ffe7144 100644 --- a/internal/volition/store.go +++ b/internal/volition/store.go @@ -724,6 +724,16 @@ func (s *Store) Apply(ctx context.Context, r Recorder, mind, id ids.ID, rec Reco if err != nil { return it, err } + return s.applyFrom(ctx, r, it, rec, t) +} + +// applyFrom applies a reconsideration to the intention as the caller read +// it. The status change is a compare-and-set against that status: two +// reconsiderations racing on one intention (the executor and the event bus, +// say) cannot both apply, and a closed intention is never reopened. A copy +// the database has moved past gets ErrClosed and writes nothing. +func (s *Store) applyFrom(ctx context.Context, r Recorder, it Intention, rec Reconsideration, t Trigger) (Intention, error) { + mind, id := it.MindID, it.ID if !it.Status.Open() { return it, fmt.Errorf("%w (%s)", ErrClosed, it.Status) } @@ -760,6 +770,20 @@ func (s *Store) Apply(ctx context.Context, r Recorder, mind, id ids.ID, rec Reco if !t.EventID.IsZero() { evidence = append(evidence, events.EvidenceRef{Kind: "event", ID: string(t.EventID), Note: "trigger: " + t.Kind}) } + // The transition is written before its event so that a lost race + // (zero rows) records nothing at all. + tag, err := r.Q.Exec(ctx, `UPDATE intentions SET status=$3, blocked_kind=$4, blocked_reason=$5, require_approval=$6, kind=$7, reconsiderations=reconsiderations+1, status_reason=$8, closed_at=coalesce($9, closed_at), updated_at=$10 WHERE id=$1 AND mind_id=$2 AND status=$11`, + id, mind, string(next), blockedKind, trunc(blockedReason, 500), requireApproval, string(kind), trunc(rec.Reason, 500), closed, now, string(it.Status)) + if err != nil { + return it, err + } + if tag.RowsAffected() == 0 { + cur, gerr := s.GetIntention(ctx, r.Q, mind, id) + if gerr != nil { + return it, gerr + } + return cur, fmt.Errorf("%w: it moved from %s to %s meanwhile", ErrClosed, it.Status, cur.Status) + } rr := r if !t.EventID.IsZero() { rr.Cause = t.EventID @@ -769,8 +793,7 @@ func (s *Store) Apply(ctx context.Context, r Recorder, mind, id ids.ID, rec Reco if err != nil { return it, err } - if _, err := r.Q.Exec(ctx, `UPDATE intentions SET status=$3, blocked_kind=$4, blocked_reason=$5, require_approval=$6, kind=$7, reconsiderations=reconsiderations+1, status_reason=$8, last_event=$9, closed_at=coalesce($10, closed_at), updated_at=$11 WHERE id=$1 AND mind_id=$2`, - id, mind, string(next), blockedKind, trunc(blockedReason, 500), requireApproval, string(kind), trunc(rec.Reason, 500), nullID(evID), closed, now); err != nil { + if _, err := r.Q.Exec(ctx, `UPDATE intentions SET last_event=$3 WHERE id=$1 AND mind_id=$2`, id, mind, nullID(evID)); err != nil { return it, err } if _, err := s.RecordDecision(ctx, r.Q, Decision{MindID: mind, Kind: "reconsideration", Outcome: rec.Outcome, Reason: t.Kind, Rationale: rec.Reason, IntentionID: id, EventID: evID, CreatedAt: now, @@ -802,12 +825,16 @@ func (s *Store) Supersede(ctx context.Context, r Recorder, mind, old, by ids.ID, return nil } now := r.now() + tag, err := r.Q.Exec(ctx, `UPDATE intentions SET status='superseded', superseded_by=$3, status_reason=$4, closed_at=$5, updated_at=$5 WHERE id=$1 AND mind_id=$2 AND status=$6`, old, mind, by, reason, now, string(it.Status)) + if err != nil || tag.RowsAffected() == 0 { + return err // moved on meanwhile: whoever closed it recorded why + } evID, err := r.emit(ctx, mind, events.IntentionSuperseded, "intention superseded: "+it.Description+" — "+reason, 0.4, []events.EvidenceRef{{Kind: "intention", ID: string(old)}, {Kind: "intention", ID: string(by), Note: "replacement"}}, map[string]any{"intention_id": string(old), "superseded_by": string(by)}, map[string]any{"reason": reason}) if err != nil { return err } - _, err = r.Q.Exec(ctx, `UPDATE intentions SET status='superseded', superseded_by=$3, status_reason=$4, last_event=$5, closed_at=$6, updated_at=$6 WHERE id=$1 AND mind_id=$2`, old, mind, by, reason, nullID(evID), now) + _, err = r.Q.Exec(ctx, `UPDATE intentions SET last_event=$3 WHERE id=$1 AND mind_id=$2`, old, mind, nullID(evID)) return err } diff --git a/internal/volition/store_test.go b/internal/volition/store_test.go index 97ae6ca..eb43eed 100644 --- a/internal/volition/store_test.go +++ b/internal/volition/store_test.go @@ -135,3 +135,40 @@ func TestStoreLifecycle(t *testing.T) { t.Fatalf("conflict with a satisfied want still open: %+v", cs) } } + +// Regression: two reconsiderations racing on one intention must not both +// apply. A caller holding a copy the database has moved past gets ErrClosed +// and writes neither a transition, an event nor a decision; an intention +// closed by one path is never reopened by the other. +func TestStaleIntentionCopyCannotApply(t *testing.T) { + pool := storage.TestPool(t, migrations.FS) + ctx := context.Background() + m, _ := minds.NewStore(pool).Create(ctx, minds.CreateParams{Slug: "race"}) + es := evstore.NewStore(pool, nil, nil) + s := NewStore(pool) + r := s.Direct(es, "volition.test") + it, err := s.CreateIntention(ctx, r, Intention{MindID: m.ID, Description: "fetch the status page", Kind: IntentAct, Status: IntentionBlocked, BlockedKind: BlockApproval}) + if err != nil { + t.Fatal(err) + } + stale := it // the executor's copy, read before the event bus abandoned it + if _, err := s.Apply(ctx, r, m.ID, it.ID, Reconsideration{Outcome: Abandon, Reason: "the goal it served was abandoned"}, Trigger{Kind: TriggerGoalAbandoned}); err != nil { + t.Fatal(err) + } + cur, err := s.applyFrom(ctx, r, stale, Reconsideration{Outcome: Resume, Reason: "the operator granted the requested authority"}, Trigger{Kind: TriggerRequestGranted}) + if !errors.Is(err, ErrClosed) || cur.Status != IntentionAbandoned || cur.ClosedAt == nil { + t.Fatalf("stale resume applied: %v %+v", err, cur) + } + if evs, _ := es.Query(ctx, evstore.Filter{MindID: m.ID, Types: []events.Type{events.IntentionResumed}}); len(evs) != 0 { + t.Fatal("a lost race still recorded an event") + } + if ds, _ := s.Decisions(ctx, m.ID, DecisionFilter{IntentionID: it.ID}); len(ds) != 1 { + t.Fatalf("decisions: %d, want the abandonment only", len(ds)) + } + if err := s.Supersede(ctx, r, m.ID, it.ID, ids.New(), "replaced"); err != nil { + t.Fatal(err) + } + if got, _ := s.GetIntention(ctx, nil, m.ID, it.ID); got.Status != IntentionAbandoned { + t.Fatalf("supersede relabelled a closed intention: %s", got.Status) + } +}