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)