Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 7 additions & 4 deletions internal/api/handler_purchases.go
Original file line number Diff line number Diff line change
Expand Up @@ -1388,15 +1388,18 @@ func (h *Handler) scheduleApprovedExecution(ctx context.Context, execution *conf
}

// Stamp the scheduled time and actor onto the post-CAS row, then persist.
// SavePurchaseExecution is a full-row upsert, so we set all extra fields on
// the returned (freshly-transitioned) execution to avoid overwriting fields
// a concurrent writer may have set between the CAS and here.
// The write is a full-row upsert, so the extra fields go on the returned
// (freshly-transitioned) execution, and it is guarded on "scheduled": a
// revoke that lands between the CAS and here flips the row to canceled,
// and an unguarded upsert would put "scheduled" back, so the scheduler
// would buy a purchase the user revoked (issue #835). A lost guard returns
// ErrExecutionNotInExpectedStatus, which approveWithDelay maps to a 409.
scheduledAt := time.Now().Add(delay)
updated.ScheduledExecutionAt = &scheduledAt
if actor != "" {
updated.ApprovedBy = &actor
}
if err := h.config.SavePurchaseExecution(ctx, updated); err != nil {
if err := h.config.SaveExecutionIfStatus(ctx, updated, []string{"scheduled"}); err != nil {
return nil, fmt.Errorf("failed to stamp scheduled_execution_at on execution %s: %w", execution.ExecutionID, err)
}
return updated, nil
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,84 @@
//go:build integration

package api

// Issue #835 on real PostgreSQL: approveWithDelay's CAS to "scheduled" wins,
// then a revoke cancels the row before the scheduled_execution_at stamp is
// saved. Real store and handler; the revoke is the exact store work
// revokeScheduledExecution does (CancelScheduledExecutionAtomic plus the
// suppression release, in one transaction). Nothing here reaches a cloud API.

import (
"context"
"sync"
"testing"
"time"

"github.com/LeanerCloud/cloud-commitments-platform/internal/config"
"github.com/google/uuid"
"github.com/jackc/pgx/v5"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
)

// revokeBeforeStampStore runs revoke once, ahead of the first execution save
// of either kind, so the revoke lands between the CAS and the stamp whether
// the stamp is guarded or not.
type revokeBeforeStampStore struct {
*config.PostgresStore
once sync.Once
revoke func()
}

func (s *revokeBeforeStampStore) SavePurchaseExecution(ctx context.Context, e *config.PurchaseExecution) error {
s.once.Do(s.revoke)
return s.PostgresStore.SavePurchaseExecution(ctx, e)
}

func (s *revokeBeforeStampStore) SaveExecutionIfStatus(ctx context.Context, e *config.PurchaseExecution, expected []string) error {
s.once.Do(s.revoke)
return s.PostgresStore.SaveExecutionIfStatus(ctx, e, expected)
}

// Before the fix the stamp's full-row upsert wrote "scheduled" back over the
// canceled row, with a scheduled_execution_at the scheduler then picks up: a
// purchase the user revoked for free would be bought anyway.
func TestApproveWithDelay_Postgres_RevokeBeforeStampStaysCanceled(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
t.Cleanup(cancel)
f := newCreateConcurrencyFixture(ctx, t)

row := &config.PurchaseExecution{
ExecutionID: uuid.NewString(), PlanID: f.planID, Status: "notified", StepNumber: 3, ScheduledDate: time.Now(),
}
require.NoError(t, f.store.SavePurchaseExecution(ctx, row))
revokedBy := "revoker@example.com"
var canceled bool
f.handler.config = &revokeBeforeStampStore{PostgresStore: f.store, revoke: func() {
require.NoError(t, f.store.WithTx(ctx, func(tx pgx.Tx) error {
var err error
if canceled, _, err = f.store.CancelScheduledExecutionAtomic(ctx, tx, row.ExecutionID, &revokedBy); err != nil {
return err
}
return f.store.DeleteSuppressionsByExecutionTx(ctx, tx, row.ExecutionID)
}))
}}

// A delay already in the past: had the stamp landed, the scheduler's due
// query would return the row on its next tick.
_, err := f.handler.approveWithDelay(ctx, row, -time.Minute, "approver@example.com", nil, "")

require.True(t, canceled, "the revoke must have won its CAS on the scheduled row")
got, getErr := f.store.GetExecutionByID(ctx, row.ExecutionID)
require.NoError(t, getErr)
assert.Equal(t, config.StatusCanceled, got.Status, "the revoke must survive the approve's stamp")
assert.Nil(t, got.ScheduledExecutionAt)
due, dueErr := f.store.GetScheduledExecutionsDue(ctx)
require.NoError(t, dueErr)
for i := range due {
assert.NotEqual(t, row.ExecutionID, due[i].ExecutionID, "a revoked row must never be due for the scheduler")
}
ce, ok := IsClientError(err)
require.True(t, ok, "expected a client error, got %v", err)
assert.Equal(t, 409, ce.code)
}
34 changes: 32 additions & 2 deletions internal/api/handler_purchases_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -5697,11 +5697,13 @@ func TestHandler_scheduleApprovedExecution_HappyPath(t *testing.T) {
mockConfig := new(MockConfigStore)
mockConfig.On("TransitionExecutionStatus", ctx, execID, []string{"pending", "notified"}, "scheduled", mock.Anything).
Return(transitioned, nil)
mockConfig.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool {
// The stamp is guarded on "scheduled" (issue #835); the mock falls back to
// SavePurchaseExecution without a guarded expectation, so this pins it.
mockConfig.On("SaveExecutionIfStatus", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool {
return e.ExecutionID == execID &&
e.ScheduledExecutionAt != nil &&
*e.ApprovedBy == "actor@example.com"
})).Return(nil)
}), []string{"scheduled"}).Return(nil)

handler := &Handler{config: mockConfig}

Expand Down Expand Up @@ -5737,6 +5739,34 @@ func TestApproveWithDelay_CASLostMaps409(t *testing.T) {
mockConfig.AssertExpectations(t)
}

// TestApproveWithDelay_StampLostToRevokeMaps409 covers issue #835: the CAS to
// scheduled wins, then a revoke flips the row to canceled before the
// scheduled_execution_at stamp. The stamp is guarded on "scheduled", so it
// writes nothing and the approver gets the same 409 as a lost CAS, instead of
// an unguarded upsert putting "scheduled" back for the scheduler to buy.
func TestApproveWithDelay_StampLostToRevokeMaps409(t *testing.T) {
ctx := context.Background()
execID := "exec-stamp-409"
exec := &config.PurchaseExecution{ExecutionID: execID, Status: "pending"}
transitioned := &config.PurchaseExecution{ExecutionID: execID, Status: "scheduled"}
revoked := fmt.Errorf("%w: execution %s is %q, expected one of [scheduled]",
config.ErrExecutionNotInExpectedStatus, execID, config.StatusCanceled)

mockConfig := new(MockConfigStore)
mockConfig.On("TransitionExecutionStatus", ctx, execID, []string{"pending", "notified"}, "scheduled", mock.Anything).
Return(transitioned, nil)
mockConfig.On("SaveExecutionIfStatus", ctx, transitioned, []string{"scheduled"}).Return(revoked).Once()

handler := &Handler{config: mockConfig}

_, err := handler.approveWithDelay(ctx, exec, 48*time.Hour, "actor@example.com", nil, "")
ce, ok := IsClientError(err)
require.True(t, ok, "a stamp lost to a revoke must be a ClientError, got %v", err)
assert.Equal(t, 409, ce.code)
mockConfig.AssertExpectations(t)
mockConfig.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything)
}

// ---------------------------------------------------------------------------
// revokePurchase handler tests (issue #291)
// ---------------------------------------------------------------------------
Expand Down
22 changes: 14 additions & 8 deletions internal/purchase/approvals.go
Original file line number Diff line number Diff line change
Expand Up @@ -389,7 +389,8 @@ func (m *Manager) RunPlannedPurchaseNow(ctx context.Context, executionID, actor
// rotation lives here, in the one funnel all approve paths share.
//
// A non-empty skipReason stamps the executed_* audit fields once the CAS wins.
// The returned row is nil when the CAS was never won.
// The returned row is nil when the CAS was never won, or when the audit stamp
// found the claim already lost (nothing is bought then; issue #835).
//
// check runs against the row the CAS returned, which is the row
// executeAndFinalize buys, before any audit stamp or provider call (issue
Expand Down Expand Up @@ -437,11 +438,10 @@ func (m *Manager) transitionApproveAndExecute(ctx context.Context, executionID,
updated.ApprovedBy = &a
}
if actor != "" || skipReason != "" {
if saveErr := m.config.SaveExecutionIfStatus(ctx, updated, executorOwnedStatuses); saveErr != nil {
// Attribution is best-effort once the atomic flip has landed --
// dropping these audit fields must not stop the purchase from firing.
// Log loudly so the audit gap is visible.
logging.Errorf("AUDIT GAP: failed to stamp approval audit fields on %s: %v", executionID, saveErr)
// Guarded on "running". A store error stays best-effort, but a lost
// guard means the claim is gone, so nothing is bought (issue #835).
if stampErr := m.stampClaimedAudit(ctx, updated, "approval audit fields"); stampErr != nil {
return nil, "", stampErr
}
}

Expand Down Expand Up @@ -568,7 +568,10 @@ func (m *Manager) rotateApprovalToken(ctx context.Context, exec *config.Purchase
expiry := time.Now().Add(ttl)
exec.ApprovalToken = config.HashApprovalToken(tok)
exec.ApprovalTokenExpiresAt = &expiry
if saveErr := m.config.SavePurchaseExecution(ctx, exec); saveErr != nil {
// Guarded on the terminal status the executor just wrote, so this
// full-row write cannot put back a status another writer set since
// (issue #835).
if saveErr := m.config.SaveExecutionIfStatus(ctx, exec, []string{exec.Status}); saveErr != nil {
return "", saveErr
}
return tok, nil
Expand All @@ -592,7 +595,10 @@ func (m *Manager) clearApprovalToken(ctx context.Context, executionID string) er
}
exec.ApprovalToken = ""
exec.ApprovalTokenExpiresAt = nil
return m.config.SavePurchaseExecution(ctx, exec)
// Guarded on "canceled": the row was read after the cancel landed, and
// this full-row write must not undo a status change made since (issue
// #835).
return m.config.SaveExecutionIfStatus(ctx, exec, []string{config.StatusCanceled})
}

// loadCancelableExecution fetches an execution, validates the approval
Expand Down
12 changes: 8 additions & 4 deletions internal/purchase/approvals_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -1333,9 +1333,12 @@ func TestClearApprovalToken_PersistsEmpty(t *testing.T) {
ApprovalTokenExpiresAt: &future,
}
store.On("GetExecutionByID", ctx, "exec-tok-clear").Return(execution, nil)
store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool {
// Guarded on "canceled" (issue #835). The mock falls back to
// SavePurchaseExecution only when no guarded expectation is set, so this
// expectation is what pins the guard.
store.On("SaveExecutionIfStatus", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool {
return e.ExecutionID == "exec-tok-clear" && e.ApprovalToken == "" && e.ApprovalTokenExpiresAt == nil
})).Return(nil)
}), []string{"canceled"}).Return(nil)

err := manager.clearApprovalToken(ctx, "exec-tok-clear")
require.NoError(t, err)
Expand All @@ -1355,13 +1358,14 @@ func TestMintRevocationToken_PersistsFreshToken(t *testing.T) {
Status: "completed",
ApprovalToken: "old-approval-token",
}
store.On("SavePurchaseExecution", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool {
// Guarded on the terminal status the row already carries (issue #835).
store.On("SaveExecutionIfStatus", ctx, mock.MatchedBy(func(e *config.PurchaseExecution) bool {
// Fresh token must be non-empty, different from old, and expiry set.
return e.ExecutionID == "exec-mint" &&
e.ApprovalToken != "" &&
e.ApprovalToken != "old-approval-token" &&
e.ApprovalTokenExpiresAt != nil
})).Return(nil)
}), []string{"completed"}).Return(nil)

rawToken, err := manager.rotateApprovalToken(ctx, execution, config.RevocationWindow)
require.NoError(t, err)
Expand Down
31 changes: 31 additions & 0 deletions internal/purchase/claim.go
Original file line number Diff line number Diff line change
Expand Up @@ -59,3 +59,34 @@ func (m *Manager) claimAndCheck(ctx context.Context, executionID string, from []
}
return updated, nil
}

// errClaimLostBeforePurchase marks a run that stopped before any provider call
// because its post-claim audit stamp found the row no longer "running": another
// writer moved it on (or it vanished) after the claim, so this run no longer
// owns it and must not buy (issue #835). It wraps the store's lost-guard error,
// so errors.Is(err, config.ErrExecutionNotInExpectedStatus) (or ErrNotFound)
// still holds for callers that map those to a 409 or a race loss.
var errClaimLostBeforePurchase = errors.New("claim lost before the purchase; nothing was bought")

// stampClaimedAudit persists the audit fields (approved_by, executed_at, ...)
// the caller set on its freshly claimed row, guarded on "running" so a stale
// copy never rewrites a status another writer set.
//
// Losing the guard is fail-closed: it returns errClaimLostBeforePurchase and
// the caller must stop before the provider call, because the guard has just
// proved the claim is gone (issue #835). Any other save error stays
// best-effort, as before: the claim is still this run's, the terminal save is
// guarded on "running" too, and missing attribution must not block a purchase.
// what names the fields for the AUDIT GAP log line.
func (m *Manager) stampClaimedAudit(ctx context.Context, exec *config.PurchaseExecution, what string) error {
err := m.config.SaveExecutionIfStatus(ctx, exec, executorOwnedStatuses)
if err == nil {
return nil
}
if errors.Is(err, config.ErrExecutionNotInExpectedStatus) || errors.Is(err, config.ErrNotFound) {
logging.Warnf("purchase[%s]: %s stamp found the claim gone; not purchasing: %v", exec.ExecutionID, what, err)
return fmt.Errorf("%w: %w", errClaimLostBeforePurchase, err)
}
logging.Errorf("AUDIT GAP: failed to stamp %s on %s: %v", what, exec.ExecutionID, err)
return nil
}
10 changes: 6 additions & 4 deletions internal/purchase/claim_guard_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -107,12 +107,14 @@ func TestApprove_ClaimsRunningAndStampsGuarded(t *testing.T) {
stubExecuteChain(t, store, sender, "plan-g")
store.On("SaveExecutionIfStatus", ctx, mock.Anything, []string{"running"}).Return(nil).Once()
store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{"running"}).Return(nil).Once()
store.On("SaveExecutionIfStatus", mock.Anything, mock.Anything, []string{"completed"}).Return(nil).Once()

_, _, err := manager.ApproveAndExecute(ctx, "exec-g", "approver@example.invalid", nil, nil)
require.NoError(t, err)
store.AssertCalled(t, "TransitionExecutionStatus", ctx, "exec-g", approveFromStatuses, "running", (*string)(nil))
// Two guarded writes (audit stamp, terminal save); the only unguarded one is
// the post-execution revocation-token rotation.
store.AssertNumberOfCalls(t, "SaveExecutionIfStatus", 2)
store.AssertNumberOfCalls(t, "SavePurchaseExecution", 1)
// Three guarded writes: the audit stamp and the terminal save on running,
// and the post-execution revocation-token rotation on the terminal status
// it follows (issue #835). No unguarded full-row write remains.
store.AssertNumberOfCalls(t, "SaveExecutionIfStatus", 3)
store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything)
}
8 changes: 6 additions & 2 deletions internal/purchase/manager.go
Original file line number Diff line number Diff line change
Expand Up @@ -386,7 +386,9 @@ func (m *Manager) recordRampAdvanceRefusal(ctx context.Context, exec *config.Pur
}
logging.Errorf("Failed to update plan progress: %v", cause)
exec.Error = appendErrNote(exec.Error, fmt.Sprintf("ramp not advanced: %v", cause))
if saveErr := m.config.SavePurchaseExecution(ctx, exec); saveErr != nil {
// Guarded on the terminal status this run just wrote, so the full-row
// write cannot put back a status another writer set since (issue #835).
if saveErr := m.config.SaveExecutionIfStatus(ctx, exec, []string{exec.Status}); saveErr != nil {
logging.Errorf("AUDIT LOSS: failed to persist ramp-advance refusal for execution %s: %v",
exec.ExecutionID, saveErr)
}
Expand Down Expand Up @@ -683,7 +685,9 @@ func (m *Manager) safeFail(ctx context.Context, exec *config.PurchaseExecution)
}

updated.Error = strandedRecoveryError(exec)
if saveErr := m.config.SavePurchaseExecution(ctx, updated); saveErr != nil {
if saveErr := m.config.SaveExecutionIfStatus(ctx, updated, []string{config.StatusFailed}); saveErr != nil {
// Guarded on "failed" (issue #835): the stamp lands only while the row
// is still in the status this CAS set, never over a later writer's.
// The atomic flip to "failed" already landed via TransitionExecutionStatus;
// only the explanatory error string failed to persist. Log loudly but
// still count the recovery - the row is no longer stranded in "approved".
Expand Down
6 changes: 5 additions & 1 deletion internal/purchase/notifications.go
Original file line number Diff line number Diff line change
Expand Up @@ -333,7 +333,11 @@ func (m *Manager) failPlanStepTx(ctx context.Context, tx pgx.Tx, exec *config.Pu
return nil, err
}
markFailedPlanStep(failed, reason)
if err := m.config.SavePurchaseExecutionTx(ctx, tx, failed); err != nil {
// The save is guarded on "failed", the status this CAS set (issue #835).
// The CAS's row lock already keeps other writers out until commit, so this
// is defense in depth: the full-row write never puts back a status another
// writer set, and a lost guard rolls back the CAS and the release with it.
if err := m.config.SaveExecutionIfStatusTx(ctx, tx, failed, []string{config.StatusFailed}); err != nil {
return nil, fmt.Errorf("failed to record plan step %d of plan %s as failed: %w", failed.StepNumber, failed.PlanID, err)
}
// The row is failed with NoSubmission, so its suppressions are released as
Expand Down
Loading
Loading