From b00e31dbff17b6f3e36e41a49dca50362219aa08 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Mon, 28 Sep 2026 02:44:09 +0200 Subject: [PATCH] fix(api): make Run now actually execute the purchase runPlannedPurchase only CASed a pending/paused execution to "running" and reported "Purchase execution initiated". No component consumes "running" rows: the scheduler only picks up pending/notified, and purchase.Reaper flips anything stuck in running for 10 minutes to failed. The operator was told the purchase ran, nothing was bought, and the row read "failed" ten minutes later. Extract ApproveAndExecute's 4-eyes-gated, CAS-then-execute body into transitionApproveAndExecute, parameterized on the source statuses, and add purchase.Manager.RunPlannedPurchaseNow (CAS from pending/paused, same as before) that runs the purchase synchronously through the same funnel ApproveAndExecute and the direct-execute path already use. runPlannedPurchase now delegates to it instead of doing a bare status flip. This makes "Run now" idempotent by construction: a second click, or the scheduler's claimAndExecute racing the same pending row, loses the same atomic UPDATE ... WHERE status IN (...) CAS the approve path already relies on, so only one caller ever executes the purchase. It also closes a real gap: forcing an execution previously bypassed the 4-eyes dual-control check entirely, unlike every other execute path. Closes #218 Co-Authored-By: claude-flow --- internal/api/handler_purchases.go | 17 ++++++--- internal/api/handler_purchases_test.go | 23 +++++++----- internal/api/handler_test.go | 10 +++-- internal/api/mocks_test.go | 5 +++ internal/api/types.go | 5 +++ internal/purchase/approvals.go | 50 ++++++++++++++++++++----- internal/purchase/approvals_test.go | 51 ++++++++++++++++++++++++++ internal/server/interfaces.go | 5 +++ internal/testutil/mocks.go | 8 ++++ 9 files changed, 146 insertions(+), 28 deletions(-) diff --git a/internal/api/handler_purchases.go b/internal/api/handler_purchases.go index 87eeb3aeb..df8957e3a 100644 --- a/internal/api/handler_purchases.go +++ b/internal/api/handler_purchases.go @@ -299,6 +299,15 @@ func (h *Handler) resumePlannedPurchase(ctx context.Context, req *events.LambdaF return &StatusResponse{Status: "resumed"}, nil } +// runPlannedPurchase forces a pending or paused scheduled purchase to +// execute immediately (the "Run now" button) instead of waiting for its +// scheduled date. It delegates to purchase.Manager.RunPlannedPurchaseNow, +// which shares ApproveAndExecute's 4-eyes-gated, CAS-guarded +// claim-then-execute funnel (issue #218): the previous implementation only +// CASed the row's status to "running" and reported success from that bare +// flip, but no executor consumes "running" rows -- the scheduler only picks +// up pending/notified, so nothing ever bought the purchase and the reaper +// later failed the row out from under the operator who was told it ran. func (h *Handler) runPlannedPurchase(ctx context.Context, req *events.LambdaFunctionURLRequest, executionID string) (any, error) { if err := validateUUID(executionID); err != nil { return nil, err @@ -315,16 +324,14 @@ func (h *Handler) runPlannedPurchase(ctx context.Context, req *events.LambdaFunc return nil, err } - // Atomically transition to running — only one concurrent caller can succeed. - // TransitionExecutionStatus handles not-found and wrong-status cases. - if _, err := h.config.TransitionExecutionStatus(ctx, executionID, []string{"pending", "paused"}, "running", resolveCreatorUserID(session)); err != nil { + if err := h.purchase.RunPlannedPurchaseNow(ctx, executionID, fourEyesActorIdentity(session), resolveCreatorUserID(session)); err != nil { return nil, NewClientError(409, fmt.Sprintf("execution %s cannot be started: %v", executionID, err)) } return map[string]any{ "execution_id": executionID, - "status": "running", - "message": "Purchase execution initiated", + "status": "completed", + "message": "Purchase executed", }, nil } diff --git a/internal/api/handler_purchases_test.go b/internal/api/handler_purchases_test.go index fbd7e2e10..a2fb0abf1 100644 --- a/internal/api/handler_purchases_test.go +++ b/internal/api/handler_purchases_test.go @@ -1246,22 +1246,22 @@ func TestHandler_runPlannedPurchase(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) mockAuth := new(MockAuthService) + mockPurchase := new(MockPurchaseManager) adminSession := &Session{ UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", Email: "admin@example.com", } - transitioned := &config.PurchaseExecution{ - ExecutionID: "11111111-1111-1111-1111-111111111111", - Status: "running", - } - mockAuth.On("ValidateSession", ctx, "admin-token").Return(adminSession, nil) mockAuth.grantAdminPurchaser() - mockStore.On("TransitionExecutionStatus", ctx, "11111111-1111-1111-1111-111111111111", []string{"pending", "paused"}, "running", mock.Anything).Return(transitioned, nil) + // runPlannedPurchase ("Run now") must delegate to the same 4-eyes-gated, + // CAS-guarded funnel ApproveAndExecute uses (issue #218) rather than a + // bare TransitionExecutionStatus flip to "running" that no executor + // consumes. + mockPurchase.On("RunPlannedPurchaseNow", ctx, "11111111-1111-1111-1111-111111111111", "admin@example.com", mock.Anything).Return(nil) - handler := &Handler{config: mockStore, auth: mockAuth} + handler := &Handler{purchase: mockPurchase, config: mockStore, auth: mockAuth} req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ @@ -1273,7 +1273,8 @@ func TestHandler_runPlannedPurchase(t *testing.T) { resultMap := result.(map[string]interface{}) assert.Equal(t, "11111111-1111-1111-1111-111111111111", resultMap["execution_id"]) - assert.Equal(t, "running", resultMap["status"]) + assert.Equal(t, "completed", resultMap["status"]) + mockPurchase.AssertExpectations(t) } func TestHandler_deletePlannedPurchase(t *testing.T) { @@ -1758,6 +1759,7 @@ func TestHandler_runPlannedPurchase_NilExecution(t *testing.T) { ctx := context.Background() mockStore := new(MockConfigStore) mockAuth := new(MockAuthService) + mockPurchase := new(MockPurchaseManager) adminSession := &Session{ UserID: "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", @@ -1766,9 +1768,10 @@ func TestHandler_runPlannedPurchase_NilExecution(t *testing.T) { mockAuth.On("ValidateSession", ctx, "admin-token").Return(adminSession, nil) mockAuth.grantAdmin() - mockStore.On("TransitionExecutionStatus", ctx, "99999999-9999-9999-9999-999999999999", []string{"pending", "paused"}, "running", mock.Anything).Return(nil, fmt.Errorf("execution not found: 99999999-9999-9999-9999-999999999999")) + mockPurchase.On("RunPlannedPurchaseNow", ctx, "99999999-9999-9999-9999-999999999999", "admin@example.com", mock.Anything). + Return(fmt.Errorf("execution not found: 99999999-9999-9999-9999-999999999999")) - handler := &Handler{config: mockStore, auth: mockAuth} + handler := &Handler{purchase: mockPurchase, config: mockStore, auth: mockAuth} req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ diff --git a/internal/api/handler_test.go b/internal/api/handler_test.go index be086679c..27e14b1e2 100644 --- a/internal/api/handler_test.go +++ b/internal/api/handler_test.go @@ -1253,10 +1253,14 @@ func TestHandler_HandleRequest_RunPlannedPurchase(t *testing.T) { mockAuth.grantAdmin() mockAuth.On("ValidateCSRFToken", ctx, mock.Anything, mock.Anything).Return(nil) - transitioned := &config.PurchaseExecution{ExecutionID: "11111111-1111-1111-1111-111111111111", Status: "running"} - mockStore.On("TransitionExecutionStatus", mock.Anything, "11111111-1111-1111-1111-111111111111", []string{"pending", "paused"}, "running", mock.Anything).Return(transitioned, nil) + mockPurchase := new(MockPurchaseManager) + // This request authenticates via the admin API key (X-API-Key), which + // resolves to the stateless admin-api-key principal rather than the + // bearer-token session, so the actor identity fourEyesActorIdentity + // derives is the sentinel, not the session's email. + mockPurchase.On("RunPlannedPurchaseNow", mock.Anything, "11111111-1111-1111-1111-111111111111", "admin-api-key", mock.Anything).Return(nil) - handler := &Handler{config: mockStore, auth: mockAuth, corsAllowedOrigin: "*", apiKey: "test-key"} + handler := &Handler{purchase: mockPurchase, config: mockStore, auth: mockAuth, corsAllowedOrigin: "*", apiKey: "test-key"} req := &events.LambdaFunctionURLRequest{ Headers: map[string]string{ diff --git a/internal/api/mocks_test.go b/internal/api/mocks_test.go index 15e1b5d13..bba502ad9 100644 --- a/internal/api/mocks_test.go +++ b/internal/api/mocks_test.go @@ -58,6 +58,11 @@ func (m *MockPurchaseManager) ApproveAndExecute(ctx context.Context, execID, act return args.Error(0) } +func (m *MockPurchaseManager) RunPlannedPurchaseNow(ctx context.Context, execID, actor string, transitionedBy *string) error { + args := m.Called(ctx, execID, actor, transitionedBy) + return args.Error(0) +} + func (m *MockPurchaseManager) CancelExecution(ctx context.Context, execID, token, actor string) error { args := m.Called(ctx, execID, token, actor) return args.Error(0) diff --git a/internal/api/types.go b/internal/api/types.go index d192f4743..6096fb846 100644 --- a/internal/api/types.go +++ b/internal/api/types.go @@ -137,6 +137,11 @@ type BreakdownValue struct { type PurchaseManagerInterface interface { ApproveExecution(ctx context.Context, execID, token, actor string) error ApproveAndExecute(ctx context.Context, execID, actor string, transitionedBy *string) error + // RunPlannedPurchaseNow forces a pending or paused scheduled purchase to + // execute immediately (the "Run now" button), sharing ApproveAndExecute's + // 4-eyes-gated, CAS-guarded funnel instead of a bare status flip that + // nothing else consumes (issue #218). + RunPlannedPurchaseNow(ctx context.Context, execID, actor string, transitionedBy *string) error CancelExecution(ctx context.Context, execID, token, actor string) error } diff --git a/internal/purchase/approvals.go b/internal/purchase/approvals.go index 20b13d585..f1be17fcd 100644 --- a/internal/purchase/approvals.go +++ b/internal/purchase/approvals.go @@ -311,17 +311,47 @@ func (m *Manager) checkDifferentApprover(ctx context.Context, executionID string // approval drives its own executeAndFinalize, which already fans out // per-account in parallel via executeMultiAccount. func (m *Manager) ApproveAndExecute(ctx context.Context, executionID, actor string, transitionedBy *string) error { + return m.transitionApproveAndExecute(ctx, executionID, actor, transitionedBy, []string{"pending", "notified"}) +} + +// RunPlannedPurchaseNow lets an operator force a scheduled purchase (pending +// or paused) to execute immediately instead of waiting for its scheduled +// date, going through the exact same 4-eyes gate and synchronous +// executeAndFinalize funnel ApproveAndExecute uses (issue #218). The +// previous runPlannedPurchase implementation only flipped the row's status +// to "running" and returned success: no executor consumes "running" rows +// (the scheduler picks up pending/notified, the reaper only cleans up a +// stuck "running" row after 10 minutes), so nothing ever bought the +// purchase and the row later reads "failed" out from under the operator who +// was told it ran. +// +// Sharing transitionApproveAndExecute's CAS with ApproveAndExecute and +// claimAndExecute is also what makes this idempotent: a second "Run now" +// click, or the scheduler racing the same pending row, loses the atomic +// UPDATE ... WHERE status IN (...) and gets a clean "cannot transition" +// error rather than executing the purchase a second time. +func (m *Manager) RunPlannedPurchaseNow(ctx context.Context, executionID, actor string, transitionedBy *string) error { + return m.transitionApproveAndExecute(ctx, executionID, actor, transitionedBy, []string{"pending", "paused"}) +} + +// transitionApproveAndExecute is the shared body behind ApproveAndExecute +// and RunPlannedPurchaseNow: enforce the 4-eyes gate, atomically CAS the row +// from one of fromStatuses to "approved", stamp ApprovedBy, then run the +// purchase synchronously. Parameterizing on fromStatuses lets both entry +// points share one implementation of the gate and the CAS instead of the +// gate being reimplemented (or skipped) at a second call site. +func (m *Manager) transitionApproveAndExecute(ctx context.Context, executionID, actor string, transitionedBy *string, fromStatuses []string) error { t0 := time.Now() - logging.Infof("purchase[%s]: ApproveAndExecute starting (actor=%q)", executionID, maskActor(actor)) + logging.Infof("purchase[%s]: transitionApproveAndExecute starting (actor=%q, from=%v)", executionID, maskActor(actor), fromStatuses) // Universal 4-eyes gate (issue #1005 / PR #1500 adversarial review): runs // before any state mutation so every caller -- session approve, direct - // execute, token approve, SQS approve -- is covered by one policy check. - // transitionedBy doubles as the actor's own UUID for this check when the - // caller has one (session-based callers); see enforceFourEyesPolicy's doc - // comment for the full rationale. + // execute, run-now, token approve, SQS approve -- is covered by one + // policy check. transitionedBy doubles as the actor's own UUID for this + // check when the caller has one (session-based callers); see + // enforceFourEyesPolicy's doc comment for the full rationale. if err := m.enforceFourEyesPolicy(ctx, executionID, actor, transitionedBy); err != nil { - logging.Warnf("purchase[%s]: ApproveAndExecute denied by 4-eyes policy: %v", executionID, err) + logging.Warnf("purchase[%s]: transitionApproveAndExecute denied by 4-eyes policy: %v", executionID, err) return err } @@ -329,9 +359,9 @@ func (m *Manager) ApproveAndExecute(ctx context.Context, executionID, actor stri // approvals (stamped onto transitioned_by); it is nil for token/SQS/system // flows so transitioned_by = NULL on those hops. The human-readable actor // email is recorded separately onto approved_by (below). - updated, err := m.config.TransitionExecutionStatus(ctx, executionID, []string{"pending", "notified"}, "approved", transitionedBy) + updated, err := m.config.TransitionExecutionStatus(ctx, executionID, fromStatuses, "approved", transitionedBy) if err != nil { - logging.Errorf("purchase[%s]: ApproveAndExecute status transition failed after %s: %v", + logging.Errorf("purchase[%s]: transitionApproveAndExecute status transition failed after %s: %v", executionID, time.Since(t0), err) return fmt.Errorf("approve: %w", err) } @@ -351,9 +381,9 @@ func (m *Manager) ApproveAndExecute(ctx context.Context, executionID, actor stri logging.Infof("purchase[%s]: executing purchase synchronously", executionID) execErr := m.executeAndFinalize(ctx, updated) if execErr != nil { - logging.Errorf("purchase[%s]: ApproveAndExecute failed after %s: %v", executionID, time.Since(t0), execErr) + logging.Errorf("purchase[%s]: transitionApproveAndExecute failed after %s: %v", executionID, time.Since(t0), execErr) } else { - logging.Infof("purchase[%s]: ApproveAndExecute completed in %s", executionID, time.Since(t0)) + logging.Infof("purchase[%s]: transitionApproveAndExecute completed in %s", executionID, time.Since(t0)) } return execErr } diff --git a/internal/purchase/approvals_test.go b/internal/purchase/approvals_test.go index 8c158ef29..6954533c6 100644 --- a/internal/purchase/approvals_test.go +++ b/internal/purchase/approvals_test.go @@ -306,6 +306,57 @@ func TestManager_ApproveAndExecute_SkipsTokenCheck(t *testing.T) { sender.AssertExpectations(t) } +// TestManager_RunPlannedPurchaseNow_ExecutesFromPaused pins issue #218: +// "Run now" on a scheduled purchase must actually execute it, not just flip +// its status. Pre-fix, the handler CASed the row straight to "running" with +// no executor consuming that state -- the scheduler only picks up +// pending/notified, so the purchase was never bought and the reaper +// eventually failed the row out from under the operator who was told it +// ran. This drives RunPlannedPurchaseNow from "paused" (a source +// ApproveAndExecute itself does not accept) and asserts the full +// executeAndFinalize chain runs: the plan lookup, the confirmation email, +// SavePurchaseExecution, and CompletePlanStep all fire, proving the +// purchase was actually executed rather than left stuck in "running". +func TestManager_RunPlannedPurchaseNow_ExecutesFromPaused(t *testing.T) { + ctx := context.Background() + manager, store, sender := newApproveManager(t) + + updated := &config.PurchaseExecution{ + ExecutionID: "exec-run-now", + PlanID: "plan-run-now", + Status: "approved", + StepNumber: 1, + } + store.On("TransitionExecutionStatus", ctx, "exec-run-now", []string{"pending", "paused"}, "approved", (*string)(nil)).Return(updated, nil) + stubExecuteChain(t, store, sender, "plan-run-now") + + err := manager.RunPlannedPurchaseNow(ctx, "exec-run-now", "operator@example.com", nil) + require.NoError(t, err) + require.NotNil(t, updated.ApprovedBy) + assert.Equal(t, "operator@example.com", *updated.ApprovedBy) + store.AssertExpectations(t) + sender.AssertExpectations(t) +} + +// TestManager_RunPlannedPurchaseNow_LostCASReturnsError proves a lost CAS +// (row already claimed -- by a concurrent "Run now" click, or the scheduler +// racing the same pending row) surfaces as an error rather than silently +// reporting success, and that no execution work runs on that path (no +// unexpected mock calls -- store has no execute-chain stubs registered). +// This is the idempotency half of issue #218's "no double purchase" ask. +func TestManager_RunPlannedPurchaseNow_LostCASReturnsError(t *testing.T) { + ctx := context.Background() + manager, store, _ := newApproveManager(t) + + store.On("TransitionExecutionStatus", ctx, "exec-race", []string{"pending", "paused"}, "approved", (*string)(nil)). + Return(nil, config.ErrExecutionNotInExpectedStatus) + + err := manager.RunPlannedPurchaseNow(ctx, "exec-race", "operator@example.com", nil) + require.Error(t, err) + assert.ErrorIs(t, err, config.ErrExecutionNotInExpectedStatus) + store.AssertExpectations(t) +} + // ─── enforceFourEyesPolicy at the ApproveAndExecute choke point ─────────────── // (issue #1005 / PR #1500 adversarial review) // diff --git a/internal/server/interfaces.go b/internal/server/interfaces.go index e99ea3f1d..ac5f5c0cb 100644 --- a/internal/server/interfaces.go +++ b/internal/server/interfaces.go @@ -27,6 +27,11 @@ type PurchaseManagerInterface interface { ProcessMessage(ctx context.Context, body string) error ApproveExecution(ctx context.Context, execID, token, actor string) error ApproveAndExecute(ctx context.Context, execID, actor string, transitionedBy *string) error + // RunPlannedPurchaseNow forces a pending or paused scheduled purchase to + // execute immediately (the "Run now" button), sharing ApproveAndExecute's + // 4-eyes-gated, CAS-guarded funnel instead of a bare status flip that + // nothing else consumes (issue #218). + RunPlannedPurchaseNow(ctx context.Context, execID, actor string, transitionedBy *string) error CancelExecution(ctx context.Context, execID, token, actor string) error // ReapStuckExecutions sweeps purchase_executions stuck in // approved/running longer than reapAfter and flips them to "failed" diff --git a/internal/testutil/mocks.go b/internal/testutil/mocks.go index ee331aba6..bae456c15 100644 --- a/internal/testutil/mocks.go +++ b/internal/testutil/mocks.go @@ -44,6 +44,7 @@ type MockPurchaseManager struct { ProcessMessageFunc func(ctx context.Context, body string) error ApproveExecutionFunc func(ctx context.Context, execID, token, actor string) error ApproveAndExecuteFunc func(ctx context.Context, execID, actor string, transitionedBy *string) error + RunPlannedPurchaseNowFunc func(ctx context.Context, execID, actor string, transitionedBy *string) error CancelExecutionFunc func(ctx context.Context, execID, token, actor string) error ReapStuckExecutionsFunc func(ctx context.Context, reapAfter time.Duration) (*purchase.ReapResult, error) FireScheduledDelayedPurchasesFunc func(ctx context.Context) (*purchase.FireResult, error) @@ -85,6 +86,13 @@ func (m *MockPurchaseManager) ApproveAndExecute(ctx context.Context, execID, act return nil } +func (m *MockPurchaseManager) RunPlannedPurchaseNow(ctx context.Context, execID, actor string, transitionedBy *string) error { + if m.RunPlannedPurchaseNowFunc != nil { + return m.RunPlannedPurchaseNowFunc(ctx, execID, actor, transitionedBy) + } + return nil +} + func (m *MockPurchaseManager) CancelExecution(ctx context.Context, execID, token, actor string) error { if m.CancelExecutionFunc != nil { return m.CancelExecutionFunc(ctx, execID, token, actor)