From 494ae7c26642e5e6efa8e2c147e16e838286c2d4 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Tue, 13 Oct 2026 00:05:07 +0200 Subject: [PATCH] feat(purchase): make notified real and pin the scheduler claim to the gated snapshot Refs #631 (PR2b-2a). Nothing attaches recommendations to plan steps yet. - RotatePendingApprovalToken now also promotes pending to notified in the same statement, so the status only changes after the notification email went out; a failed send leaves the row pending and the scheduler refuses it. - claimAndExecute claims with TransitionExecutionStatusPinned, which requires the row's recommendations to still be empty (or non-empty) as the gate saw them. A lost pin is a benign skip on the cron and SQS paths. - A claimed pending/notified row re-runs CheckStepCommitmentCap against the current plan and fails with the beforeSubmission class so suppressions release; approved and web rows are unchanged. - planstep.Precheck / checkStoredSPCount fail closed on Include*/Exclude*/ MinCount filters, an immediate ramp with percent_per_step != 100 and a Savings Plans stored row with count != 1; Resolve calls them first. Co-Authored-By: Claude Sonnet 5.5 --- internal/analytics/collector_test.go | 4 + internal/config/interfaces.go | 6 + internal/config/store_postgres.go | 21 +- internal/config/store_postgres_db_test.go | 64 ++++- .../config/store_postgres_pgxmock_test.go | 2 +- internal/mocks/stores.go | 20 ++ internal/planstep/precheck.go | 71 ++++++ internal/planstep/precheck_test.go | 86 +++++++ internal/planstep/resolve.go | 6 + internal/planstep/resolve_test.go | 3 +- internal/purchase/claim_pin_test.go | 225 ++++++++++++++++++ internal/purchase/manager.go | 25 +- .../notified_claim_pin_integration_test.go | 129 ++++++++++ internal/purchase/step_cap.go | 26 ++ internal/server/test_helpers_test.go | 4 + 15 files changed, 682 insertions(+), 10 deletions(-) create mode 100644 internal/planstep/precheck.go create mode 100644 internal/planstep/precheck_test.go create mode 100644 internal/purchase/claim_pin_test.go create mode 100644 internal/purchase/notified_claim_pin_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 8b5e0017..208e1758 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -303,6 +303,10 @@ func (m *mockConfigStore) TransitionExecutionStatus(ctx context.Context, executi return nil, nil } +func (m *mockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) { + return nil, nil +} + func (m *mockConfigStore) SetCancelledBy(_ context.Context, _ string, _ string) error { return nil } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 7f6b64fd..e19d57cc 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -177,6 +177,12 @@ type StoreInterface interface { // When non-nil the actor is stamped onto transitioned_by + transitioned_at; when nil, // transitioned_by is set to NULL and transitioned_at is still set to NOW() for ordering. TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*PurchaseExecution, error) + // TransitionExecutionStatusPinned is TransitionExecutionStatus that also + // requires the row's recommendations to be empty (sawEmptyRecommendations) + // or non-empty exactly as the caller saw them, so a claim cannot land on a + // row whose recommendations were attached after the caller's gate check. A + // lost pin returns ErrExecutionNotInExpectedStatus like any lost CAS. + TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) // SetCancelledBy stamps canceled_by and the legacy cancelled_by column on // an execution without overwriting any other columns. Used after // TransitionExecutionStatus to fold the actor attribution into the same diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index d92acdad..44293613 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -1168,12 +1168,15 @@ func (s *PostgresStore) LinkRetryExecutionAtomic(ctx context.Context, tx pgx.Tx, return tag.RowsAffected() == 1, nil } -// RotatePendingApprovalToken implements StoreInterface. +// RotatePendingApprovalToken implements StoreInterface. The same statement +// that stores the emailed token promotes the row to 'notified' (a no-op for a +// row already there), so the status never claims a notice that was not sent. func (s *PostgresStore) RotatePendingApprovalToken(ctx context.Context, executionID, tokenHash string, expiresAt time.Time) (bool, error) { q := ` UPDATE purchase_executions SET approval_token_hash = $2, approval_token_expires_at = $3, + status = 'notified', updated_at = NOW() WHERE execution_id = $1 AND status IN ('pending', 'notified') @@ -1194,6 +1197,19 @@ func (s *PostgresStore) RotatePendingApprovalToken(ctx context.Context, executio // and the legacy cancelled_by (issue #119), matching what the other cancel paths // record and what History reads. func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string) (*PurchaseExecution, error) { + return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, nil) +} + +// TransitionExecutionStatusPinned is TransitionExecutionStatus that also +// requires the row's recommendations to still be empty (or not empty), as the +// caller saw them. A recommendation attach landing between a gate check and the +// claim makes the pin fail with ErrExecutionNotInExpectedStatus, so a row is +// never run on a snapshot different from the one that was gated. +func (s *PostgresStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*PurchaseExecution, error) { + return s.transitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor, &sawEmptyRecommendations) +} + +func (s *PostgresStore) transitionExecutionStatus(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmpty *bool) (*PurchaseExecution, error) { var cancelActor *string if toStatus == StatusCanceled { cancelActor = actor @@ -1205,6 +1221,7 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution canceled_by = COALESCE((SELECT email FROM users WHERE id = $5), canceled_by), cancelled_by = COALESCE((SELECT email FROM users WHERE id = $5), cancelled_by) WHERE execution_id = $1 AND status = ANY($3) + AND ($6::boolean IS NULL OR (recommendations IN ('[]'::jsonb, 'null'::jsonb)) = $6) RETURNING plan_id, execution_id, status, step_number, scheduled_date, notification_sent, approval_token_hash, recommendations, total_upfront_cost, estimated_savings, completed_at, error, expires_at, @@ -1215,7 +1232,7 @@ func (s *PostgresStore) TransitionExecutionStatus(ctx context.Context, execution idempotency_key, scheduled_execution_at ` - records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor) + records, err := s.queryExecutions(ctx, query, executionID, toStatus, fromStatuses, actor, cancelActor, sawEmpty) if err != nil { return nil, err } diff --git a/internal/config/store_postgres_db_test.go b/internal/config/store_postgres_db_test.go index 79fab266..b24e15d1 100644 --- a/internal/config/store_postgres_db_test.go +++ b/internal/config/store_postgres_db_test.go @@ -1661,7 +1661,8 @@ func TestPostgresStoreDB_LinkRetryExecutionAtomic(t *testing.T) { // TestPostgresStoreDB_RotatePendingApprovalToken (issue #103): the // notification path's token rotation writes approval_token_hash and its -// expiry only while the row is pending/notified, and touches nothing else +// expiry, and promotes the row to notified, only while it is pending/notified, +// and touches nothing else // (a full-row upsert of a stale copy could revert a concurrent approve). // The hash round-trips through the approval_token_hash column. func TestPostgresStoreDB_RotatePendingApprovalToken(t *testing.T) { @@ -1707,7 +1708,7 @@ func TestPostgresStoreDB_RotatePendingApprovalToken(t *testing.T) { assert.Equal(t, HashApprovalToken("fresh-token"), reread.ApprovalToken) require.NotNil(t, reread.ApprovalTokenExpiresAt) assert.True(t, expiresAt.Equal(reread.ApprovalTokenExpiresAt.UTC().Truncate(time.Second))) - assert.Equal(t, status, reread.Status, "rotation must not touch status") + assert.Equal(t, "notified", reread.Status, "the token rotation is also the pending->notified CAS") }) } @@ -1720,6 +1721,65 @@ func TestPostgresStoreDB_RotatePendingApprovalToken(t *testing.T) { reread, err := store.GetExecutionByID(ctx, exec.ExecutionID) require.NoError(t, err) assert.Equal(t, HashApprovalToken("original-token"), reread.ApprovalToken) + assert.Equal(t, "completed", reread.Status, "a row that left pending must not be revived to notified") + }) +} + +// TestPostgresStoreDB_TransitionExecutionStatusPinned: the claim only wins +// while the row's recommendations are still as the caller saw them, so a +// recommendation attach between a gate check and the claim loses the pin and +// leaves the row untouched. +func TestPostgresStoreDB_TransitionExecutionStatusPinned(t *testing.T) { + conn := setupTestContainerDB(t) + if conn == nil { + return + } + store := NewPostgresStore(conn) + ctx := context.Background() + plan := &PurchasePlan{Name: "pinned claim", Enabled: true, Services: map[string]ServiceConfig{}, RampSchedule: PresetRampSchedules["immediate"]} + require.NoError(t, store.CreatePurchasePlan(ctx, plan)) + rec := []RecommendationRecord{{ID: "r1", Provider: "aws", Service: "ec2", Count: 1}} + newRow := func(t *testing.T, recs []RecommendationRecord) *PurchaseExecution { + t.Helper() + exec := &PurchaseExecution{PlanID: plan.ID, Status: "notified", StepNumber: 1, ScheduledDate: time.Now(), Recommendations: recs} + require.NoError(t, store.SavePurchaseExecution(ctx, exec)) + return exec + } + from := []string{"approved", "pending", "notified"} + + for _, empty := range []([]RecommendationRecord){nil, {}} { + exec := newRow(t, empty) + _, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, false) + require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus, "gate saw recommendations but the row is empty") + reread, err := store.GetExecutionByID(ctx, exec.ExecutionID) + require.NoError(t, err) + assert.Equal(t, "notified", reread.Status) + got, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, true) + require.NoError(t, err) + assert.Equal(t, "running", got.Status) + } + + t.Run("attach after the gate saw an empty row loses the claim", func(t *testing.T) { + exec := newRow(t, nil) + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + ok, err := store.SetExecutionRecommendationsIfEmptyTx(ctx, tx, exec.ExecutionID, rec, 1, 1) + require.True(t, ok) + return err + })) + _, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, true) + require.ErrorIs(t, err, ErrExecutionNotInExpectedStatus) + reread, err := store.GetExecutionByID(ctx, exec.ExecutionID) + require.NoError(t, err) + assert.Equal(t, "notified", reread.Status) + got, err := store.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, from, "running", nil, false) + require.NoError(t, err) + assert.Equal(t, "running", got.Status, "the next tick gates the attached row and claims it") + }) + + t.Run("unpinned transition is unchanged", func(t *testing.T) { + got, err := store.TransitionExecutionStatus(ctx, newRow(t, rec).ExecutionID, from, "running", nil) + require.NoError(t, err) + assert.Equal(t, "running", got.Status) }) } diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index f51017bc..ed27dc91 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -2773,7 +2773,7 @@ func TestPGXMock_TransitionExecutionStatus_ProbeHardErrorNotMappedToNotFound(t * // CAS UPDATE matches zero rows (status already transitioned or row gone). mock.ExpectQuery(`UPDATE purchase_executions`). - WithArgs(anyArgsCfg(5)...). + WithArgs(anyArgsCfg(6)...). WillReturnRows(pgxmock.NewRows(execCols)) // Probe fails with a hard DB error, not an empty result. diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 1296446a..e92eb4e3 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -357,6 +357,26 @@ func (m *MockConfigStore) TransitionExecutionStatus(ctx context.Context, executi return v, args.Error(1) } +// TransitionExecutionStatusPinned mocks the pinned claim. With no expectation +// registered it delegates to TransitionExecutionStatus, so tests that mock the +// plain transition keep working; a test exercising a lost pin registers an +// expectation returning config.ErrExecutionNotInExpectedStatus. +func (m *MockConfigStore) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) { + if !isExpected(&m.Mock, "TransitionExecutionStatusPinned") { + return m.TransitionExecutionStatus(ctx, executionID, fromStatuses, toStatus, actor) + } + m.record("TransitionExecutionStatusPinned", ctx, executionID, fromStatuses, toStatus, actor, sawEmptyRecommendations) + args := m.Called(ctx, executionID, fromStatuses, toStatus, actor, sawEmptyRecommendations) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).(*config.PurchaseExecution) + if !ok { + panic(fmt.Sprintf("mock: expected *config.PurchaseExecution, got %T", args.Get(0))) + } + return v, args.Error(1) +} + // SetCancelledBy mocks the SetCancelledBy targeted-update operation. func (m *MockConfigStore) SetCancelledBy(ctx context.Context, executionID, cancelledBy string) error { m.record("SetCancelledBy", ctx, executionID, cancelledBy) diff --git a/internal/planstep/precheck.go b/internal/planstep/precheck.go new file mode 100644 index 00000000..1926d051 --- /dev/null +++ b/internal/planstep/precheck.go @@ -0,0 +1,71 @@ +package planstep + +import ( + "errors" + "fmt" + "sort" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// ErrUnsupportedPlan marks a plan whose settings Resolve cannot honor. Every +// such plan is refused before any recommendation is read, rather than resolved +// with the setting silently ignored. +var ErrUnsupportedPlan = errors.New("plan settings are not supported by recommendation resolution") + +// Precheck refuses, with a plan-specific error, the plan settings Resolve does +// not apply: any Include*/Exclude*/MinCount filter on a service config (the +// plan-step request never sets them, so honoring them would buy something other +// than what the plan says), and an immediate ramp whose PercentPerStep is not +// 100 (an immediate ramp ignores that field when validated but Resolve sizes +// with it). +func Precheck(plan *config.PurchasePlan) error { + keys := make([]string, 0, len(plan.Services)) + for k := range plan.Services { + keys = append(keys, k) + } + sort.Strings(keys) + for _, k := range keys { + svc := plan.Services[k] + if field := firstFilterSet(&svc); field != "" { + return fmt.Errorf("%w: plan %s service %q sets %s", ErrUnsupportedPlan, plan.ID, k, field) + } + } + if plan.RampSchedule.Type == "immediate" && plan.RampSchedule.PercentPerStep != 100 { + return fmt.Errorf("%w: plan %s has an immediate ramp with percent_per_step %v (must be 100)", + ErrUnsupportedPlan, plan.ID, plan.RampSchedule.PercentPerStep) + } + return nil +} + +// firstFilterSet names the first filter field set on svc, or "". +func firstFilterSet(svc *config.ServiceConfig) string { + switch { + case len(svc.IncludeEngines) > 0: + return "include_engines" + case len(svc.ExcludeEngines) > 0: + return "exclude_engines" + case len(svc.IncludeRegions) > 0: + return "include_regions" + case len(svc.ExcludeRegions) > 0: + return "exclude_regions" + case len(svc.IncludeTypes) > 0: + return "include_types" + case len(svc.ExcludeTypes) > 0: + return "exclude_types" + case svc.MinCount != 0: + return "min_count" + } + return "" +} + +// checkStoredSPCount refuses a Savings Plans stored row whose Count is not 1: +// an SP is sized by its hourly commitment, so any other count means the row is +// not the one-commitment shape the sizing assumes. +func checkStoredSPCount(stored *config.RecommendationRecord) error { + if common.IsSavingsPlan(common.ServiceType(stored.Service)) && stored.Count != 1 { + return fmt.Errorf("%w: savings plan stored row %s has count %d (must be 1)", ErrUnsupportedPlan, stored.ID, stored.Count) + } + return nil +} diff --git a/internal/planstep/precheck_test.go b/internal/planstep/precheck_test.go new file mode 100644 index 00000000..00fdf03e --- /dev/null +++ b/internal/planstep/precheck_test.go @@ -0,0 +1,86 @@ +package planstep + +import ( + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// Fixtures are hand-built plans, not real configuration. + +func planWith(svc config.ServiceConfig, ramp config.RampSchedule) *config.PurchasePlan { + return &config.PurchasePlan{ID: "plan-pc", Services: map[string]config.ServiceConfig{"aws:ec2": svc}, RampSchedule: ramp} +} + +func TestPrecheck(t *testing.T) { + ok := config.ServiceConfig{Provider: "aws", Service: "ec2", Enabled: true} + weekly := config.RampSchedule{Type: "weekly", PercentPerStep: 25} + cases := []struct { + name string + plan *config.PurchasePlan + want string // substring of the error; "" = accepted + }{ + {"clean plan", planWith(ok, weekly), ""}, + {"immediate at 100", planWith(ok, config.RampSchedule{Type: "immediate", PercentPerStep: 100}), ""}, + {"immediate at 50", planWith(ok, config.RampSchedule{Type: "immediate", PercentPerStep: 50}), "immediate ramp"}, + {"immediate at 0", planWith(ok, config.RampSchedule{Type: "immediate"}), "immediate ramp"}, + {"weekly at 25 is not an immediate ramp", planWith(ok, weekly), ""}, + {"include engines", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.IncludeEngines = []string{"mysql"} }), weekly), "include_engines"}, + {"exclude engines", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.ExcludeEngines = []string{"mysql"} }), weekly), "exclude_engines"}, + {"include regions", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.IncludeRegions = []string{"us-east-1"} }), weekly), "include_regions"}, + {"exclude regions", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.ExcludeRegions = []string{"us-east-1"} }), weekly), "exclude_regions"}, + {"include types", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.IncludeTypes = []string{"m5.large"} }), weekly), "include_types"}, + {"exclude types", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.ExcludeTypes = []string{"m5.large"} }), weekly), "exclude_types"}, + {"min count", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.MinCount = 1 }), weekly), "min_count"}, + {"negative min count", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.MinCount = -1 }), weekly), "min_count"}, + {"filter on a disabled service still refuses", planWith(withSvc(ok, func(s *config.ServiceConfig) { s.Enabled = false; s.MinCount = 2 }), weekly), "min_count"}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + err := Precheck(tc.plan) + if tc.want == "" { + assert.NoError(t, err) + return + } + require.ErrorIs(t, err, ErrUnsupportedPlan) + assert.ErrorContains(t, err, tc.want) + assert.ErrorContains(t, err, "plan-pc") + }) + } +} + +func withSvc(s config.ServiceConfig, f func(*config.ServiceConfig)) config.ServiceConfig { + f(&s) + return s +} + +func TestCheckStoredSPCount(t *testing.T) { + sp := func(count int) *config.RecommendationRecord { + return &config.RecommendationRecord{ID: "r", Service: "savings-plans-compute", Count: count} + } + assert.NoError(t, checkStoredSPCount(sp(1))) + for _, n := range []int{0, 2, -1} { + require.ErrorIs(t, checkStoredSPCount(sp(n)), ErrUnsupportedPlan, "count %d", n) + } + // A count-denominated service is not constrained. + assert.NoError(t, checkStoredSPCount(&config.RecommendationRecord{Service: "ec2", Count: 7})) +} + +func TestResolveRefusesUnsupportedPlanBeforeReading(t *testing.T) { + plan := azurePlan(100, 100) + svc := plan.Services["azure:compute"] + svc.MinCount = 3 + plan.Services["azure:compute"] = svc + // A nil Source/Store would panic if Resolve read anything. + _, err := Resolve(t.Context(), nil, nil, plan, accts) + require.ErrorIs(t, err, ErrUnsupportedPlan) +} + +func TestResolveRefusesStoredSPWithCountNotOne(t *testing.T) { + stored := spRec(`{"hourly_commitment":1.234}`) + stored.Count = 3 + _, err := Resolve(t.Context(), fakeSource{shown: []config.RecommendationRecord{stored}}, fakeStore{rows: []config.RecommendationRecord{stored}}, spPlan(80, 25), accts) + require.ErrorIs(t, err, ErrUnsupportedPlan) +} diff --git a/internal/planstep/resolve.go b/internal/planstep/resolve.go index 71a03bdf..08401775 100644 --- a/internal/planstep/resolve.go +++ b/internal/planstep/resolve.go @@ -46,6 +46,9 @@ type Store interface { // excluded. Every failure is returned; there is no unsized or over-show // fallback. An empty result is ErrNoMatch. func Resolve(ctx context.Context, src Source, store Store, plan *config.PurchasePlan, accountIDs []string) ([]config.RecommendationRecord, error) { + if err := Precheck(plan); err != nil { + return nil, err + } if len(accountIDs) == 0 { return nil, fmt.Errorf("plan %s has no target accounts", plan.ID) } @@ -213,6 +216,9 @@ func size(shown, stored *config.RecommendationRecord, product float64) (config.R if stored.Count <= 0 { return config.RecommendationRecord{}, false, fmt.Errorf("stored count %d cannot price a share", stored.Count) } + if err := checkStoredSPCount(stored); err != nil { + return config.RecommendationRecord{}, false, err + } out := *shown out.SuppressedCount, out.SuppressionExpiresAt, out.PrimarySuppressionExecutionID = 0, nil, nil out.Selected = true diff --git a/internal/planstep/resolve_test.go b/internal/planstep/resolve_test.go index 1bd05338..a1b6f51e 100644 --- a/internal/planstep/resolve_test.go +++ b/internal/planstep/resolve_test.go @@ -216,9 +216,8 @@ func TestResolveSavingsPlanRoundsDownAndKeepsRawKeys(t *testing.T) { stored := spRec(`{"hourly_commitment":1.234,"platform":"Linux/UNIX","tenancy":"default","plan_type":"Compute","extra":{"a":[1,2]}}`) before, err := stored.ComputeIdentity() require.NoError(t, err) - stored.Count = 3 + stored.Count = 1 shown := stored - shown.Count = 2 // partially suppressed placeholder count got, err := Resolve(t.Context(), fakeSource{shown: []config.RecommendationRecord{shown}}, fakeStore{rows: []config.RecommendationRecord{stored}}, spPlan(80, 25), accts) require.NoError(t, err) require.Len(t, got, 1) diff --git a/internal/purchase/claim_pin_test.go b/internal/purchase/claim_pin_test.go new file mode 100644 index 00000000..6bf41fe2 --- /dev/null +++ b/internal/purchase/claim_pin_test.go @@ -0,0 +1,225 @@ +package purchase + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// Fixtures below are MockConfigStore expectations and hand-built records, not +// real cloud data. A MockProviderFactory with no expectations panics on any +// provider call, so "no provider call" is asserted by the absence of a panic +// plus an explicit AssertNotCalled. + +var errPinLost = errors.New("pin lost") + +// newPinManager wires store for one due plan row on the cron and SQS paths. The +// plan the gate reads is gatePlan; every later GetPurchasePlan read (the +// post-claim re-check) returns claimPlan. +func newPinManager(ctx context.Context, store *MockConfigStore, factory *MockProviderFactory, exec config.PurchaseExecution, gatePlan, claimPlan *config.PurchasePlan) *Manager { + store.On("GetStaleApprovedExecutions", ctx, mock.Anything).Return([]config.PurchaseExecution{}, nil) + store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{exec}, nil) + store.On("GetExecutionByID", ctx, exec.ExecutionID).Return(&exec, nil) + store.On("GetPurchasePlan", ctx, exec.PlanID).Return(gatePlan, nil).Once() + store.On("GetPurchasePlan", ctx, exec.PlanID).Return(claimPlan, nil) + store.On("GetGlobalConfig", ctx).Return(autoPurchaseOnCfg(), nil) + return &Manager{config: store, email: new(MockEmailSender), providerFactory: factory} +} + +func autoPlan(c *float64) *config.PurchasePlan { + return &config.PurchasePlan{ID: "plan-gate", AutoPurchase: true, MaxStepCommitment: c} +} + +// S2: an attach landing between the gate (which saw an empty row) and the claim +// makes the pinned claim lose; nothing runs, on both the cron and SQS paths. +func TestClaimPin_AttachBetweenGateAndClaimIsRefused(t *testing.T) { + ctx := context.Background() + for _, path := range []string{"cron", "sqs"} { + t.Run(path, func(t *testing.T) { + store, factory := new(MockConfigStore), new(MockProviderFactory) + manager := newPinManager(ctx, store, factory, dueRecExec("notified"), autoPlan(f64(1000)), autoPlan(f64(1000))) + store.On("TransitionExecutionStatusPinned", ctx, "exec-gate", []string{"approved", "pending", "notified"}, "running", (*string)(nil), true). + Return(nil, config.ErrExecutionNotInExpectedStatus) + + if path == "cron" { + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 0, result.Executed) + assert.Equal(t, 0, result.Failed, "a lost pin is a benign skip") + } else { + require.NoError(t, manager.handleExecutePurchase(ctx, &AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: "exec-gate"})) + } + + store.AssertCalled(t, "TransitionExecutionStatusPinned", ctx, "exec-gate", []string{"approved", "pending", "notified"}, "running", (*string)(nil), true) + store.AssertNotCalled(t, "TransitionExecutionStatus", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) + store.AssertNotCalled(t, "SavePurchaseExecution", mock.Anything, mock.Anything) + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + }) + } +} + +// A row the gate saw WITH recommendations is claimed with the opposite pin. +func TestClaimPin_RecommendationBearingRowPinsNonEmpty(t *testing.T) { + ctx := context.Background() + store, factory := new(MockConfigStore), new(MockProviderFactory) + manager := newPinManager(ctx, store, factory, dueRecExec("notified", upfrontRec(10)), autoPlan(f64(1000)), autoPlan(f64(1000))) + store.On("TransitionExecutionStatusPinned", ctx, "exec-gate", mock.Anything, "running", (*string)(nil), false). + Return(nil, config.ErrExecutionNotInExpectedStatus) + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 0, result.Failed) + store.AssertCalled(t, "TransitionExecutionStatusPinned", ctx, "exec-gate", mock.Anything, "running", (*string)(nil), false) + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) +} + +// A real error from the pinned claim is a failure, not a skip. +func TestClaimPin_ClaimErrorIsFailure(t *testing.T) { + ctx := context.Background() + store, factory := new(MockConfigStore), new(MockProviderFactory) + manager := newPinManager(ctx, store, factory, dueRecExec("notified"), autoPlan(f64(1000)), autoPlan(f64(1000))) + store.On("TransitionExecutionStatusPinned", ctx, "exec-gate", mock.Anything, "running", (*string)(nil), true).Return(nil, errPinLost) + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 1, result.Failed) +} + +// S2: the cap is re-applied to the claimed row against the plan as it is now. +// The gate saw cap 1000 and the plan was then lowered to 5: the claimed row is +// failed with the beforeSubmission class (suppressions released), no provider +// call is made. +func TestClaimPin_PostClaimCapRecheckFailsBeforeSubmission(t *testing.T) { + ctx := context.Background() + for _, path := range []string{"cron", "sqs"} { + t.Run(path, func(t *testing.T) { + exec := dueRecExec("notified", upfrontRec(10)) + exec.Recommendations[0].ID = "rec-1" + store, factory := new(MockConfigStore), new(MockProviderFactory) + manager := newPinManager(ctx, store, factory, exec, autoPlan(f64(1000)), autoPlan(f64(5))) + claimed := exec + claimed.Status = "running" + store.On("TransitionExecutionStatusPinned", ctx, "exec-gate", mock.Anything, "running", (*string)(nil), false).Return(&claimed, nil) + var saved *config.PurchaseExecution + store.On("SavePurchaseExecutionTx", ctx, mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { saved = args.Get(2).(*config.PurchaseExecution) }).Return(nil) + store.On("DeleteSuppressionsByExecutionTx", ctx, mock.Anything, "exec-gate").Return(nil) + + if path == "cron" { + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 1, result.Failed) + assert.Equal(t, 0, result.Executed) + } else { + err := manager.handleExecutePurchase(ctx, &AsyncMessage{Type: MessageTypeExecutePurchase, ExecutionID: "exec-gate"}) + require.ErrorIs(t, err, ErrStepCommitmentCap) + } + + require.NotNil(t, saved) + assert.Equal(t, "failed", saved.Status) + assert.Contains(t, saved.Error, "exceeds max_step_commitment") + store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", ctx, mock.Anything, "exec-gate") + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) + }) + } +} + +// A post-claim plan read failure also fails the claimed row closed rather than +// running it unchecked. +func TestClaimPin_PostClaimPlanReadErrorFailsClosed(t *testing.T) { + ctx := context.Background() + exec := dueRecExec("notified", upfrontRec(10)) + store := new(MockConfigStore) + store.On("GetPurchasePlan", ctx, exec.PlanID).Return(nil, errors.New("db down")) + claimed := exec + claimed.Status = "running" + store.On("TransitionExecutionStatusPinned", ctx, exec.ExecutionID, mock.Anything, "running", (*string)(nil), false).Return(&claimed, nil) + var saved *config.PurchaseExecution + store.On("SavePurchaseExecutionTx", ctx, mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { saved = args.Get(2).(*config.PurchaseExecution) }).Return(nil) + manager := &Manager{config: store, email: new(MockEmailSender), providerFactory: new(MockProviderFactory)} + + didClaim, err := manager.claimAndExecute(ctx, &exec) + require.True(t, didClaim) + require.Error(t, err) + assert.ErrorContains(t, err, "post-claim cap check") + require.NotNil(t, saved) + assert.Equal(t, "failed", saved.Status) + store.AssertCalled(t, "DeleteSuppressionsByExecutionTx", ctx, mock.Anything, exec.ExecutionID) +} + +// Manual / approved purchases are unchanged: an approved row is claimed and run +// with no cap re-check, even on a plan that has no cap (the cap error class +// never appears and the plan is read once, by executePurchase itself). +func TestClaimPin_ApprovedRowSkipsCapRecheck(t *testing.T) { + ctx := context.Background() + exec := dueRecExec("approved", upfrontRec(1e9)) + store := new(MockConfigStore) + store.On("GetPurchasePlan", ctx, exec.PlanID).Return(autoPlan(nil), nil) + store.On("GetPlanAccounts", ctx, exec.PlanID).Return(nil, errors.New("accounts unavailable")) + claimed := exec + claimed.Status = "running" + store.On("TransitionExecutionStatusPinned", ctx, exec.ExecutionID, mock.Anything, "running", (*string)(nil), false).Return(&claimed, nil) + store.On("SavePurchaseExecution", ctx, mock.Anything).Return(nil) + manager := &Manager{config: store, email: new(MockEmailSender), providerFactory: new(MockProviderFactory)} + + didClaim, err := manager.claimAndExecute(ctx, &exec) + require.True(t, didClaim) + require.Error(t, err) + assert.NotErrorIs(t, err, ErrStepCommitmentCap) + store.AssertNumberOfCalls(t, "GetPurchasePlan", 1) +} + +// S1: when the notification email fails nothing is promoted, and the scheduler +// then refuses the still-pending rec-bearing row (no claim, no failure). +func TestNotification_EmailFailureLeavesRowPendingAndSchedulerRefusesIt(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + sender := new(MockEmailSender) + plan, existing := pendingExecForNotification(store) + notify := "notify@example.com" + store.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{NotificationEmail: ¬ify, AutoPurchaseEnabled: true}, nil) + sender.On("SendScheduledPurchaseNotification", ctx, mock.Anything).Return(errors.New("ses down")) + manager := &Manager{config: store, email: sender} + + require.False(t, manager.sendPlanNotification(ctx, plan)) + store.AssertNotCalled(t, "RotatePendingApprovalToken", mock.Anything, mock.Anything, mock.Anything, mock.Anything) + + // The row as the DB still holds it: pending. The scheduler leaves it alone. + due := *existing + due.ScheduledDate = time.Now().Add(-time.Hour) + store.On("GetStaleApprovedExecutions", ctx, mock.Anything).Return([]config.PurchaseExecution{}, nil) + store.On("GetPendingExecutions", ctx).Return([]config.PurchaseExecution{due}, nil) + store.On("GetPurchasePlan", ctx, due.PlanID).Return(&config.PurchasePlan{ID: due.PlanID, AutoPurchase: true, MaxStepCommitment: f64(1000)}, nil) + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 0, result.Processed) + assert.Equal(t, 0, result.Failed) + store.AssertNotCalled(t, "TransitionExecutionStatusPinned", mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything, mock.Anything) +} + +// Rows without recommendations keep their handling (S10): the gate lets them +// through to the claim, where executePurchase fails them. Unchanged by the pin. +func TestClaimPin_RecLessRowStillFailsInsteadOfBuying(t *testing.T) { + ctx := context.Background() + store, factory := new(MockConfigStore), new(MockProviderFactory) + exec := dueRecExec("pending") + manager := newPinManager(ctx, store, factory, exec, autoPlan(nil), autoPlan(nil)) + claimed := exec + claimed.Status = "running" + store.On("TransitionExecutionStatusPinned", ctx, "exec-gate", mock.Anything, "running", (*string)(nil), true).Return(&claimed, nil) + var saved *config.PurchaseExecution + store.On("SavePurchaseExecutionTx", ctx, mock.Anything, mock.Anything). + Run(func(args mock.Arguments) { saved = args.Get(2).(*config.PurchaseExecution) }).Return(nil) + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 1, result.Failed) + require.NotNil(t, saved) + assert.Equal(t, "failed", saved.Status) + assert.Contains(t, saved.Error, ErrPlanStepNoRecommendations.Error()) + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) +} diff --git a/internal/purchase/manager.go b/internal/purchase/manager.go index e05719e5..4d18d67b 100644 --- a/internal/purchase/manager.go +++ b/internal/purchase/manager.go @@ -174,15 +174,29 @@ func (m *Manager) finalizeExecution(exec *config.PurchaseExecution, execErr erro // the caller should ack/skip without re-running the purchase. A real DB error // during the claim returns (false, err). // +// The claim is pinned to the recommendations snapshot the caller gated on: a row +// that was empty when gated must still be empty, and a row with recommendations +// must still have them. A recommendation attach landing between the gate and the +// claim therefore loses the pin, which is a benign skip (the next tick gates the +// row again), instead of running recommendations nobody checked against the cap +// or the notified rule. +// +// For a pending/notified row (the ones the AutoPurchase gate applies to) the +// step cap is re-checked against the claimed row and the current plan; a refusal +// fails the row with the beforeSubmission class, so its suppressions release. +// // On a won claim it runs executeAndFinalize and returns (true, execErr). func (m *Manager) claimAndExecute(ctx context.Context, exec *config.PurchaseExecution) (claimed bool, err error) { + gated := exec.Status == "pending" || exec.Status == "notified" + gateSawEmpty := len(exec.Recommendations) == 0 // System-initiated: SQS/cron executor passes nil so transitioned_by = NULL. - updated, claimErr := m.config.TransitionExecutionStatus(ctx, exec.ExecutionID, []string{"approved", "pending", "notified"}, "running", nil) + updated, claimErr := m.config.TransitionExecutionStatusPinned(ctx, exec.ExecutionID, []string{"approved", "pending", "notified"}, "running", nil, gateSawEmpty) if claimErr != nil { if errors.Is(claimErr, config.ErrNotFound) || errors.Is(claimErr, config.ErrExecutionNotInExpectedStatus) { // Benign CAS race-loss: another worker/redelivery already owns this - // row (or it was deleted). Ack/skip without executing. - logging.Warnf("Skipping execution %s (CAS claim lost to concurrent worker): %v", exec.ExecutionID, claimErr) + // row (or it was deleted), or its recommendations changed since the + // gate. Ack/skip without executing. + logging.Warnf("Skipping execution %s (CAS claim lost to concurrent worker or changed recommendations): %v", exec.ExecutionID, claimErr) return false, nil } return false, fmt.Errorf("failed to claim execution %s for execution: %w", exec.ExecutionID, claimErr) @@ -191,6 +205,11 @@ func (m *Manager) claimAndExecute(ctx context.Context, exec *config.PurchaseExec // the RETURNING clause) onto the caller's struct so the rest of the run // starts from the claimed row rather than the pre-claim snapshot. *exec = *updated + if gated && len(exec.Recommendations) > 0 { + if capErr := m.recheckStepCap(ctx, exec); capErr != nil { + return true, m.failClaimedBeforeSubmission(ctx, exec, capErr) + } + } return true, m.executeAndFinalize(ctx, exec) } diff --git a/internal/purchase/notified_claim_pin_integration_test.go b/internal/purchase/notified_claim_pin_integration_test.go new file mode 100644 index 00000000..d6653e19 --- /dev/null +++ b/internal/purchase/notified_claim_pin_integration_test.go @@ -0,0 +1,129 @@ +//go:build integration + +package purchase + +import ( + "context" + "errors" + "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/mock" + "github.com/stretchr/testify/require" +) + +// Real PostgreSQL, mock email sender and mock provider factory (labeled +// fixtures; nothing here reaches a cloud API). + +func claimPinFixture(ctx context.Context, t *testing.T, store *config.PostgresStore) *config.PurchasePlan { + t.Helper() + global, err := store.GetGlobalConfig(ctx) + require.NoError(t, err) + recipient := "fixture@example.invalid" + global.NotificationEmail = &recipient + global.AutoPurchaseEnabled = true + require.NoError(t, store.SaveGlobalConfig(ctx, global)) + plan := atomicStepPlan(ctx, t, store) + plan.MaxStepCommitment = f64(1000) + require.NoError(t, store.UpdatePurchasePlan(ctx, plan)) + return plan +} + +func claimPinRow(ctx context.Context, t *testing.T, store *config.PostgresStore, plan *config.PurchasePlan, status string, recs []config.RecommendationRecord) *config.PurchaseExecution { + t.Helper() + row := &config.PurchaseExecution{ + PlanID: plan.ID, ExecutionID: uuid.New().String(), Status: status, StepNumber: 1, + ScheduledDate: time.Now().Add(-time.Hour), Recommendations: recs, + } + require.NoError(t, store.SavePurchaseExecution(ctx, row)) + return row +} + +func statusInDB(ctx context.Context, t *testing.T, store *config.PostgresStore, id string) *config.PurchaseExecution { + t.Helper() + row, err := store.GetExecutionByID(ctx, id) + require.NoError(t, err) + return row +} + +// S1: a failed email leaves the row pending and the scheduler refuses it; a +// sent email makes it notified, which the scheduler gate then accepts. +func TestNotifiedTransition_OnlyAfterEmailSucceeds(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + plan := claimPinFixture(ctx, t, store) + // The notification window is open for the plan's next date (tomorrow). + row := claimPinRow(ctx, t, store, plan, "pending", []config.RecommendationRecord{upfrontRec(10)}) + factory := new(MockProviderFactory) + + sender := new(MockEmailSender) + sender.On("SendScheduledPurchaseNotification", ctx, mock.Anything).Return(errors.New("ses down")).Once() + manager := &Manager{config: store, email: sender, providerFactory: factory} + require.False(t, manager.sendPlanNotification(ctx, plan)) + assert.Equal(t, "pending", statusInDB(ctx, t, store, row.ExecutionID).Status) + + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + assert.Equal(t, 0, result.Processed, "a pending rec-bearing row is refused by the scheduler") + assert.Equal(t, "pending", statusInDB(ctx, t, store, row.ExecutionID).Status) + + sender.On("SendScheduledPurchaseNotification", ctx, mock.Anything).Return(nil).Once() + require.True(t, manager.sendPlanNotification(ctx, plan)) + notified := statusInDB(ctx, t, store, row.ExecutionID) + assert.Equal(t, "notified", notified.Status) + assert.NotEmpty(t, notified.ApprovalToken) + + // Notified now passes the scheduler gate (the claim itself is not run here). + eligible, err := manager.executableByScheduler(ctx, notified) + require.NoError(t, err) + assert.True(t, eligible) + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) +} + +// attachingStore performs a recommendation attach between the gate and the +// claim, the race S2 closes. +type attachingStore struct { + *config.PostgresStore + onClaim func() +} + +func (a *attachingStore) TransitionExecutionStatusPinned(ctx context.Context, id string, from []string, to string, actor *string, sawEmpty bool) (*config.PurchaseExecution, error) { + a.onClaim() + return a.PostgresStore.TransitionExecutionStatusPinned(ctx, id, from, to, actor, sawEmpty) +} + +// S2 on real PostgreSQL: the gate sees a rec-less notified row, an attach lands, +// the claim must refuse (row untouched, still notified, recommendations kept) +// and nothing is bought. The next tick gates the attached row with the cap and +// the notified rule. +func TestClaimPin_RealAttachBetweenGateAndClaim(t *testing.T) { + ctx := context.Background() + store, _ := rampStepStore(ctx, t) + plan := claimPinFixture(ctx, t, store) + row := claimPinRow(ctx, t, store, plan, "notified", nil) + factory := new(MockProviderFactory) + attached := false + wrapped := &attachingStore{PostgresStore: store, onClaim: func() { + attached = true + require.NoError(t, store.WithTx(ctx, func(tx pgx.Tx) error { + ok, err := store.SetExecutionRecommendationsIfEmptyTx(ctx, tx, row.ExecutionID, []config.RecommendationRecord{upfrontRec(10)}, 10, 1) + require.True(t, ok) + return err + })) + }} + manager := &Manager{config: wrapped, email: new(MockEmailSender), providerFactory: factory} + + result, err := manager.ProcessScheduledPurchases(ctx) + require.NoError(t, err) + require.True(t, attached, "the gate must have passed the rec-less row") + assert.Equal(t, 0, result.Executed) + assert.Equal(t, 0, result.Failed) + after := statusInDB(ctx, t, store, row.ExecutionID) + assert.Equal(t, "notified", after.Status, "the lost pin leaves the row alone") + assert.Len(t, after.Recommendations, 1) + factory.AssertNotCalled(t, "CreateAndValidateProvider", mock.Anything, mock.Anything, mock.Anything) +} diff --git a/internal/purchase/step_cap.go b/internal/purchase/step_cap.go index 3f0124e9..5f475798 100644 --- a/internal/purchase/step_cap.go +++ b/internal/purchase/step_cap.go @@ -1,10 +1,12 @@ package purchase import ( + "context" "errors" "fmt" "math" + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" ) @@ -43,3 +45,27 @@ func CheckStepCommitmentCap(plan *config.PurchasePlan, recs []config.Recommendat } return nil } + +// recheckStepCap re-reads the plan and re-applies CheckStepCommitmentCap to a +// row this worker has just claimed. A plan that cannot be read refuses too. +func (m *Manager) recheckStepCap(ctx context.Context, exec *config.PurchaseExecution) error { + plan, err := m.config.GetPurchasePlan(ctx, exec.PlanID) + if err != nil { + return fmt.Errorf("failed to re-read plan %s for the post-claim cap check: %w", exec.PlanID, err) + } + return CheckStepCommitmentCap(plan, exec.Recommendations) +} + +// failClaimedBeforeSubmission fails a claimed (running) row that was refused +// before any provider call, with the beforeSubmission class so its suppressions +// release. It returns the refusal, joined with ErrAuditLoss when the terminal +// save fails (the row would then be stranded in "running" for the reaper). +func (m *Manager) failClaimedBeforeSubmission(ctx context.Context, exec *config.PurchaseExecution, cause error) error { + execErr := beforeSubmission(cause) + m.finalizeExecution(exec, execErr) + logging.Errorf("Refused claimed purchase %s before submission: %v", exec.ExecutionID, cause) + if err := m.saveTerminalExecution(ctx, exec, execErr); err != nil { + return fmt.Errorf("%w: terminal save failed (%w); original refusal: %w", config.ErrAuditLoss, err, execErr) + } + return execErr +} diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 13335aea..127fef78 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -154,6 +154,10 @@ func (m *mockConfigStoreForHealth) TransitionExecutionStatus(ctx context.Context return nil, nil } +func (m *mockConfigStoreForHealth) TransitionExecutionStatusPinned(ctx context.Context, executionID string, fromStatuses []string, toStatus string, actor *string, sawEmptyRecommendations bool) (*config.PurchaseExecution, error) { + return nil, nil +} + func (m *mockConfigStoreForHealth) SetCancelledBy(_ context.Context, _ string, _ string) error { return nil }