andy/noema
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(volition): intention and plan transitions are compare-and-set

Browse Source 
Apply and Supersede updated an intention by id alone, so a reconsideration
from the event bus and one from the executor could both apply: an
intention abandoned on goal_abandoned could be resumed by resumeParent a
moment later, leaving it active with closed_at set. Each transition now
writes only when the status is still the one the caller read, before its
event and decision, and a lost race returns ErrClosed having written
nothing. Plan SetStatus likewise changes only an open plan or a draft, so
a stale copy can never relabel a completed, failed, abandoned or
superseded plan. Regression tests cover both.
This commit is contained in:
Andy Dixon
2026-10-05 19:43:22 +01:00
parent 5cf3813a2e
commit 581e9009cd
4 changed files with 98 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/planning/executor_test.go
+25
View File
@@ -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)
}
}
internal/planning/store.go
+6 -2
View File
@@ -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
}
internal/volition/store.go
+30 -3
View File
@@ -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
}
internal/volition/store_test.go
+37
View File
@@ -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)
}
}