From 766dc5381f5dfc2c816762a4f2e67ec589732410 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 15 Oct 2026 22:53:04 +0100 Subject: [PATCH 1/3] fix(purchase): pin each execution lineage's fan-out shape The single-account path derives provider idempotency tokens from the lineage key and the per-account fan-out from ":", but the path was chosen from the plan's live plan_accounts. A plan that gained its first account, or lost its last one, between an attempt and its Retry derived tokens the provider had never seen and bought again. The first attempt of a lineage now pins its shape in the new purchase_execution_fan_out_shapes table (first writer wins) and later attempts follow the pin: a single-account lineage re-drives the single-account path, and a fanned-out lineage whose plan has no accounts left is refused. Lineages that never cross keep byte-identical tokens. Migration 000123 backfills "multi" for lineages that already have a per-account row. Closes #97 Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 3 + internal/config/interfaces.go | 4 + .../config/store_postgres_fan_out_shape.go | 42 ++++ ...postgres_fan_out_shape_integration_test.go | 60 +++++ ...purchase_execution_fan_out_shapes.down.sql | 1 + ...3_purchase_execution_fan_out_shapes.up.sql | 32 +++ ..._purchase_execution_fan_out_shapes_test.go | 68 ++++++ internal/mocks/fan_out_shape.go | 43 ++++ internal/mocks/stores.go | 4 + internal/purchase/execution.go | 9 +- internal/purchase/fan_out_shape.go | 67 ++++++ internal/purchase/fan_out_shape_test.go | 207 ++++++++++++++++++ internal/server/test_helpers_test.go | 3 + 13 files changed, 541 insertions(+), 2 deletions(-) create mode 100644 internal/config/store_postgres_fan_out_shape.go create mode 100644 internal/config/store_postgres_fan_out_shape_integration_test.go create mode 100644 internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql create mode 100644 internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.up.sql create mode 100644 internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes_test.go create mode 100644 internal/mocks/fan_out_shape.go create mode 100644 internal/purchase/fan_out_shape.go create mode 100644 internal/purchase/fan_out_shape_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 45511bed..cd51efac 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -839,6 +839,9 @@ func (m *mockConfigStore) SaveExecutionIfStatus(ctx context.Context, e *config.P func (m *mockConfigStore) SaveExecutionIfStatusTx(ctx context.Context, _ pgx.Tx, e *config.PurchaseExecution, _ []string) error { return m.SavePurchaseExecution(ctx, e) } +func (m *mockConfigStore) PinExecutionFanOutShape(_ context.Context, _ string, shape string) (string, error) { + return shape, nil +} func (m *mockConfigStore) RecoverReapedExecution(ctx context.Context, e *config.PurchaseExecution) error { return m.SavePurchaseExecution(ctx, e) } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index a3013706..20484d57 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -138,6 +138,10 @@ type StoreInterface interface { // outcome over a reaper-failed row that has no retry successor linked // (ErrExecutionRetryLinked when one is). RecoverReapedExecution(ctx context.Context, execution *PurchaseExecution) error + // PinExecutionFanOutShape records shape (FanOutShapeSingle or + // FanOutShapeMulti) for the execution lineage lineageKey unless one is + // already recorded, and returns the recorded shape (issue #97). + PinExecutionFanOutShape(ctx context.Context, lineageKey, shape string) (string, error) GetPendingExecutions(ctx context.Context) ([]PurchaseExecution, error) // GetExecutionsByStatuses returns executions in any of the given states, // newest first, capped at `limit`. Used by the History handler to render diff --git a/internal/config/store_postgres_fan_out_shape.go b/internal/config/store_postgres_fan_out_shape.go new file mode 100644 index 00000000..5373dd19 --- /dev/null +++ b/internal/config/store_postgres_fan_out_shape.go @@ -0,0 +1,42 @@ +package config + +import ( + "context" + "fmt" +) + +// Fan-out shapes of a purchase execution lineage (issue #97). The shape decides +// which key the per-rec provider idempotency tokens derive from: the lineage key +// itself on the single-account path, ":" per account on +// the fan-out path. +const ( + FanOutShapeSingle = "single" + FanOutShapeMulti = "multi" +) + +// PinExecutionFanOutShape records shape as the fan-out shape of the execution +// lineage keyed by lineageKey unless one is already recorded, and returns the +// recorded shape. The first writer wins and the row is never updated, so every +// later attempt of the lineage gets back the shape its first attempt ran with, +// however the plan's accounts have changed since. +func (s *PostgresStore) PinExecutionFanOutShape(ctx context.Context, lineageKey, shape string) (string, error) { + if lineageKey == "" { + return "", fmt.Errorf("cannot pin a fan-out shape for an empty lineage key") + } + if shape != FanOutShapeSingle && shape != FanOutShapeMulti { + return "", fmt.Errorf("cannot pin unknown fan-out shape %q for lineage %s", shape, lineageKey) + } + // DO UPDATE with a no-op SET (rather than DO NOTHING) so RETURNING yields + // the stored row on conflict, in one statement that also waits out a + // concurrent first insert of the same key. + var pinned string + err := s.db.QueryRow(ctx, ` + INSERT INTO purchase_execution_fan_out_shapes (idempotency_key, shape) + VALUES ($1, $2) + ON CONFLICT (idempotency_key) DO UPDATE SET shape = purchase_execution_fan_out_shapes.shape + RETURNING shape`, lineageKey, shape).Scan(&pinned) + if err != nil { + return "", fmt.Errorf("failed to pin fan-out shape for lineage %s: %w", lineageKey, err) + } + return pinned, nil +} diff --git a/internal/config/store_postgres_fan_out_shape_integration_test.go b/internal/config/store_postgres_fan_out_shape_integration_test.go new file mode 100644 index 00000000..0ffa3612 --- /dev/null +++ b/internal/config/store_postgres_fan_out_shape_integration_test.go @@ -0,0 +1,60 @@ +//go:build integration +// +build integration + +package config + +import ( + "context" + "sync" + "testing" + + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestPinExecutionFanOutShape pins the issue #97 store contract against a real +// PostgreSQL: the first shape pinned for a lineage is returned to every later +// caller, including concurrent first callers, and invalid input is rejected. +func TestPinExecutionFanOutShape(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + t.Run("first writer wins", func(t *testing.T) { + key := uuid.NewString() + got, err := store.PinExecutionFanOutShape(ctx, key, FanOutShapeSingle) + require.NoError(t, err) + assert.Equal(t, FanOutShapeSingle, got) + + got, err = store.PinExecutionFanOutShape(ctx, key, FanOutShapeMulti) + require.NoError(t, err) + assert.Equal(t, FanOutShapeSingle, got, "a later attempt must get the first attempt's shape back") + }) + + t.Run("concurrent first attempts agree", func(t *testing.T) { + key := uuid.NewString() + shapes := []string{FanOutShapeSingle, FanOutShapeMulti, FanOutShapeSingle, FanOutShapeMulti} + got := make([]string, len(shapes)) + var wg sync.WaitGroup + for i := range shapes { + wg.Add(1) + go func(i int) { + defer wg.Done() + s, err := store.PinExecutionFanOutShape(ctx, key, shapes[i]) + assert.NoError(t, err) + got[i] = s + }(i) + } + wg.Wait() + for i := range got { + assert.Equal(t, got[0], got[i], "every concurrent caller must see the one pinned shape") + } + }) + + t.Run("invalid input", func(t *testing.T) { + _, err := store.PinExecutionFanOutShape(ctx, "", FanOutShapeSingle) + assert.Error(t, err) + _, err = store.PinExecutionFanOutShape(ctx, uuid.NewString(), "both") + assert.Error(t, err) + }) +} diff --git a/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql new file mode 100644 index 00000000..e1b97c8e --- /dev/null +++ b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql @@ -0,0 +1 @@ +DROP TABLE IF EXISTS purchase_execution_fan_out_shapes; diff --git a/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.up.sql b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.up.sql new file mode 100644 index 00000000..a92140c1 --- /dev/null +++ b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.up.sql @@ -0,0 +1,32 @@ +-- Migration 000123 (issue #97): pin the fan-out shape of each purchase +-- execution lineage. +-- +-- The per-rec provider idempotency token is derived from the lineage key on the +-- single-account path and from ":" on the per-account +-- fan-out path. The executor used to choose the path from the plan's CURRENT +-- plan_accounts, so a plan that gained its first account (or lost its last one) +-- between an attempt and its Retry derived a token the provider had never seen +-- and bought the commitment again. +-- +-- One row per lineage key (purchase_executions.idempotency_key, or the +-- execution_id for rows that predate migration 000066), written once by the +-- first execution attempt and never updated: every later attempt of the same +-- lineage takes the recorded shape instead of the live plan configuration. +CREATE TABLE IF NOT EXISTS purchase_execution_fan_out_shapes ( + idempotency_key TEXT PRIMARY KEY, + shape TEXT NOT NULL CHECK (shape IN ('single', 'multi')), + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +-- Backfill only what the data proves. A per-account row's lineage key is +-- ":" and only the fan-out path writes one, so its root +-- lineage fanned out. A root lineage with no per-account row is left unpinned +-- (its next attempt pins the live shape, the pre-#97 behavior): a fan-out +-- reaped before any per-account row was saved leaves no such row, so inferring +-- "single" from their absence could pin the wrong shape. +INSERT INTO purchase_execution_fan_out_shapes (idempotency_key, shape) +SELECT DISTINCT split_part(idempotency_key, ':', 1), 'multi' + FROM purchase_executions + WHERE idempotency_key LIKE '%:%' + AND split_part(idempotency_key, ':', 1) <> '' +ON CONFLICT (idempotency_key) DO NOTHING; diff --git a/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes_test.go b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes_test.go new file mode 100644 index 00000000..b233ad95 --- /dev/null +++ b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes_test.go @@ -0,0 +1,68 @@ +//go:build integration +// +build integration + +package migrations_test + +import ( + "context" + "os" + "path/filepath" + "testing" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/google/uuid" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestMigration_PurchaseExecutionFanOutShapes pins migration 000123 (issue #97) +// against a real PostgreSQL: a lineage with a per-account row is backfilled as +// "multi", a lineage without one is left unpinned, the up statement is +// idempotent, and the down migration drops the table. +func TestMigration_PurchaseExecutionFanOutShapes(t *testing.T) { + ctx := context.Background() + migrationsPath := getMigrationsPath() + + container, err := testhelpers.SetupPostgresContainer(ctx, t) + require.NoError(t, err) + defer container.Cleanup(ctx) + pool := container.DB.Pool() + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 123))) + + fannedOut, singleOnly := uuid.NewString(), uuid.NewString() + for _, key := range []string{fannedOut, fannedOut + ":" + uuid.NewString(), singleOnly} { + _, err = pool.Exec(ctx, ` + INSERT INTO purchase_executions (status, scheduled_date, idempotency_key) + VALUES ('failed', NOW(), $1)`, key) + require.NoError(t, err) + } + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, 123)) + + var shape string + require.NoError(t, pool.QueryRow(ctx, + `SELECT shape FROM purchase_execution_fan_out_shapes WHERE idempotency_key = $1`, fannedOut).Scan(&shape)) + assert.Equal(t, "multi", shape, "a lineage with a per-account row fanned out") + + var n int + require.NoError(t, pool.QueryRow(ctx, + `SELECT count(*) FROM purchase_execution_fan_out_shapes WHERE idempotency_key = $1`, singleOnly).Scan(&n)) + assert.Zero(t, n, "a lineage without a per-account row must not be guessed") + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM purchase_execution_fan_out_shapes`).Scan(&n)) + assert.Equal(t, 1, n, "only the root lineage is pinned, never a per-account key") + + up, err := os.ReadFile(filepath.Join(migrationsPath, "000123_purchase_execution_fan_out_shapes.up.sql")) + require.NoError(t, err) + _, err = pool.Exec(ctx, string(up)) + require.NoError(t, err, "the up statement must be idempotent") + + _, err = pool.Exec(ctx, `INSERT INTO purchase_execution_fan_out_shapes (idempotency_key, shape) VALUES ($1, 'both')`, uuid.NewString()) + assert.Error(t, err, "only single and multi are valid shapes") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, previousMigrationVersion(t, 123))) + var exists bool + require.NoError(t, pool.QueryRow(ctx, `SELECT to_regclass('purchase_execution_fan_out_shapes') IS NOT NULL`).Scan(&exists)) + assert.False(t, exists, "down migration must drop the table") +} diff --git a/internal/mocks/fan_out_shape.go b/internal/mocks/fan_out_shape.go new file mode 100644 index 00000000..1b494d84 --- /dev/null +++ b/internal/mocks/fan_out_shape.go @@ -0,0 +1,43 @@ +package mocks + +import ( + "context" + "sync" +) + +// fanOutShapePins is the in-memory stand-in for the +// purchase_execution_fan_out_shapes table: the first shape pinned for a lineage +// key is kept and returned to every later caller. The zero value is ready. +type fanOutShapePins struct { + mu sync.Mutex + pins map[string]string +} + +func (p *fanOutShapePins) pin(lineageKey, shape string) string { + p.mu.Lock() + defer p.mu.Unlock() + if pinned, ok := p.pins[lineageKey]; ok { + return pinned + } + if p.pins == nil { + p.pins = make(map[string]string) + } + p.pins[lineageKey] = shape + return shape +} + +// PinExecutionFanOutShape mocks the fan-out shape pin (issue #97). +// PinExecutionFanOutShapeFn takes priority; with no expectation registered it +// behaves like the real store (first writer wins per lineage key), so one mock +// shared across successive attempts of a lineage sees what the first pinned. +func (m *MockConfigStore) PinExecutionFanOutShape(ctx context.Context, lineageKey, shape string) (string, error) { + m.record("PinExecutionFanOutShape", ctx, lineageKey, shape) + if m.PinExecutionFanOutShapeFn != nil { + return m.PinExecutionFanOutShapeFn(ctx, lineageKey, shape) + } + if !isExpected(&m.Mock, "PinExecutionFanOutShape") { + return m.fanOutShapes.pin(lineageKey, shape), nil + } + args := m.Called(ctx, lineageKey, shape) + return args.String(0), args.Error(1) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 21803269..ebb48a84 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -52,11 +52,15 @@ type MockConfigStore struct { CountPendingExecutionsForAccountFn func(ctx context.Context, accountID string) (int, error) ListPendingExecutionIDsForAccountFn func(ctx context.Context, accountID string) ([]string, error) SavePurchaseExecutionFn func(ctx context.Context, exec *config.PurchaseExecution) error + PinExecutionFanOutShapeFn func(ctx context.Context, lineageKey, shape string) (string, error) GetUserEmailByIDFn func(ctx context.Context, userID string) (string, error) mock.Mock // callLog records every invocation, including the ones served from a // default without reaching mock.Called. See assertions.go. callLog callLog + // fanOutShapes backs PinExecutionFanOutShape's default: first writer wins + // per lineage key, like the real table. + fanOutShapes fanOutShapePins } // GetGlobalConfig mocks the GetGlobalConfig operation. Returns an empty diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 72b2e61d..60212f84 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -72,13 +72,18 @@ func (m *Manager) executePurchase(ctx context.Context, exec *config.PurchaseExec return beforeSubmission(fmt.Errorf("execution %s: %w", exec.ExecutionID, ErrPlanStepNoRecommendations)) } - // Fan out across plan accounts when accounts are configured. + // Fan out across plan accounts when accounts are configured, unless + // this lineage's first attempt pinned the other shape (issue #97). if exec.CloudAccountID == nil { accounts, err := m.config.GetPlanAccounts(ctx, exec.PlanID) if err != nil { return fmt.Errorf("failed to load plan accounts for plan %s: %w", exec.PlanID, err) } - if len(accounts) > 0 { + fanOut, err := m.lineageFansOut(ctx, exec, accounts) + if err != nil { + return err + } + if fanOut { return m.executeScopeAware(ctx, exec, plan, accounts) } } diff --git a/internal/purchase/fan_out_shape.go b/internal/purchase/fan_out_shape.go new file mode 100644 index 00000000..be46c8ff --- /dev/null +++ b/internal/purchase/fan_out_shape.go @@ -0,0 +1,67 @@ +package purchase + +import ( + "context" + "fmt" + "strings" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" +) + +// lineageFansOut decides whether a plan execution with no cloud_account_id fans +// out across the plan's accounts or runs the single-account path (issue #97). +// +// The two paths derive their provider idempotency tokens from different keys: +// the lineage key itself on the single-account path, ":" on the fan-out path. Choosing the path from the plan's CURRENT accounts +// let a plan that gained its first account, or lost its last one, between an +// attempt and its Retry derive tokens the provider had never seen, so a +// commitment the first attempt bought was bought again. Instead, the first +// attempt of a lineage pins the shape it runs with, and every later attempt +// follows the pin: +// +// - pinned shape matches the live one: unchanged behavior, so tokens of a +// lineage whose plan never crossed zero-to-N accounts are byte-identical. +// - pinned single, plan now has accounts: run the single-account path again, +// re-deriving the first attempt's tokens so the provider dedupes. +// - pinned multi, plan now has no accounts: refuse. The first attempt's tokens +// are per account and there is no account left to fan out to; the +// single-account path would derive tokens the provider has never seen. +// +// A key with a colon is a per-account lineage that reached the executor without +// its scope; reattachAccountScope (issue #1537) owns that case, so it is left to +// the live decision. An empty key cannot be pinned meaningfully and is left to +// the live decision too. +func (m *Manager) lineageFansOut(ctx context.Context, exec *config.PurchaseExecution, accounts []config.CloudAccount) (bool, error) { + live := len(accounts) > 0 + key := idempotencyLineageKey(exec) + if key == "" || strings.Contains(key, ":") { + return live, nil + } + want := config.FanOutShapeSingle + if live { + want = config.FanOutShapeMulti + } + pinned, err := m.config.PinExecutionFanOutShape(ctx, key, want) + if err != nil { + return false, fmt.Errorf("execution %s: %w", exec.ExecutionID, err) + } + switch pinned { + case want: + return live, nil + case config.FanOutShapeSingle: + logging.Warnf("purchase[%s]: plan %s now has %d account(s), but this execution's lineage first ran on the "+ + "single-account path; re-driving that path so the provider idempotency tokens match (issue #97)", + exec.ExecutionID, exec.PlanID, len(accounts)) + return false, nil + case config.FanOutShapeMulti: + return false, fmt.Errorf("refusing execution %s: its lineage first fanned out across plan %s's accounts, "+ + "which now has none, so the per-account idempotency tokens cannot be re-derived and the single-account "+ + "path would buy again under tokens the provider has never seen (issue #97). "+ + "Re-attach the accounts to the plan to let the retry complete, or cancel this execution", + exec.ExecutionID, exec.PlanID) + default: + return false, fmt.Errorf("execution %s: unknown pinned fan-out shape %q", exec.ExecutionID, pinned) + } +} diff --git a/internal/purchase/fan_out_shape_test.go b/internal/purchase/fan_out_shape_test.go new file mode 100644 index 00000000..91f51778 --- /dev/null +++ b/internal/purchase/fan_out_shape_test.go @@ -0,0 +1,207 @@ +package purchase + +import ( + "context" + "errors" + "sort" + "sync" + "testing" + + "github.com/LeanerCloud/cloud-commitments-go/pkg/common" + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// lineageRig drives successive attempts of ONE execution lineage (the first +// run and its Retry successors) through executePurchase against ONE store, so +// whatever the store persisted on the first attempt is visible to the next. +// The plan's account set is mutable between attempts, which is the +// configuration change issue #97 is about. +type lineageRig struct { + t *testing.T + store *MockConfigStore + manager *Manager + mu sync.Mutex + planAccounts []config.CloudAccount + tokens []string +} + +// shapeTestAccountA is a cloud account that exists in config throughout; only +// its membership in the plan changes between attempts. +var shapeTestAccountA = config.CloudAccount{ + ID: "acct-A", Name: "A", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys", +} + +func newLineageRig(t *testing.T) *lineageRig { + t.Helper() + r := &lineageRig{t: t, store: new(MockConfigStore)} + mockEmail := new(MockEmailSender) + mockFactory := new(MockProviderFactory) + mockProviderInst := new(MockProvider) + mockServiceClient := new(MockServiceClient) + + r.store.GetPurchasePlanFn = func(_ context.Context, id string) (*config.PurchasePlan, error) { + return &config.PurchasePlan{ID: id, Name: "Plan X"}, nil + } + r.store.GetPlanAccountsFn = func(_ context.Context, _ string) ([]config.CloudAccount, error) { + r.mu.Lock() + defer r.mu.Unlock() + return append([]config.CloudAccount(nil), r.planAccounts...), nil + } + r.store.GetCloudAccountFn = func(_ context.Context, id string) (*config.CloudAccount, error) { + if id == shapeTestAccountA.ID { + acct := shapeTestAccountA + return &acct, nil + } + return nil, nil + } + r.store.SavePurchaseExecutionFn = func(_ context.Context, _ *config.PurchaseExecution) error { return nil } + r.store.On("SavePurchaseHistory", mock.Anything, mock.AnythingOfType("*config.PurchaseHistoryRecord")).Return(nil).Maybe() + r.store.On("CompletePlanStep", mock.Anything, mock.Anything, mock.Anything).Return(nil).Maybe() + mockEmail.On("SendPurchaseConfirmation", mock.Anything, mock.AnythingOfType("email.NotificationData")).Return(nil).Maybe() + mockFactory.On("CreateAndValidateProvider", mock.Anything, "aws", mock.Anything).Return(mockProviderInst, nil).Maybe() + mockProviderInst.On("GetServiceClient", mock.Anything, common.ServiceEC2, mock.Anything).Return(mockServiceClient, nil).Maybe() + // The whole provider chain WOULD sell, so a guard that fails open shows up + // as an extra or mismatched token rather than as a mock panic. + mockServiceClient.On("PurchaseCommitment", mock.Anything, mock.Anything, mock.AnythingOfType("common.PurchaseOptions")). + Run(func(args mock.Arguments) { + r.mu.Lock() + r.tokens = append(r.tokens, args.Get(2).(common.PurchaseOptions).IdempotencyToken) + r.mu.Unlock() + }). + Return(common.PurchaseResult{Success: true, CommitmentID: "ri-ok"}, nil).Maybe() + + r.manager = &Manager{ + config: r.store, + email: mockEmail, + providerFactory: mockFactory, + credStore: awsAccessKeyCredStore(), + dashboardURL: "https://dashboard.example.com", + } + return r +} + +func (r *lineageRig) setPlanAccounts(accounts ...config.CloudAccount) { + r.mu.Lock() + defer r.mu.Unlock() + r.planAccounts = accounts +} + +// attempt runs one execution of the lineage and returns the provider tokens it +// purchased under (one entry == one real purchase) and executePurchase's error. +// Each attempt is a fresh row, as a Retry successor is: a new ExecutionID, the +// same IdempotencyKey, no cloud_account_id and unpurchased recommendations. +func (r *lineageRig) attempt(executionID string) ([]string, error) { + r.t.Helper() + r.mu.Lock() + r.tokens = nil + r.mu.Unlock() + exec := &config.PurchaseExecution{ + ExecutionID: executionID, + IdempotencyKey: "lineage-97", + PlanID: "plan-x", + StepNumber: 1, + Source: common.PurchaseSourceWeb, + Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}, + } + err := r.manager.executePurchase(context.Background(), exec) + r.mu.Lock() + defer r.mu.Unlock() + out := append([]string(nil), r.tokens...) + sort.Strings(out) + return out, err +} + +// TestRetryAfterPlanGainsFirstAccountReusesSingleAccountToken is the issue #97 +// regression guard, zero-to-N direction. +// +// The first attempt runs while the plan has no plan_accounts, so it takes the +// single-account path and buys under DeriveIdempotencyToken(K, 0). It is then +// reaped and marked failed although the commitment landed. An operator attaches +// the account to the plan and retries. The successor carries the same lineage +// key, and must buy under the SAME token so the provider dedupes; deriving the +// fan-out token DeriveIdempotencyToken(K+":acct-A", 0) instead is a token the +// provider has never seen, and the commitment is bought twice. +func TestRetryAfterPlanGainsFirstAccountReusesSingleAccountToken(t *testing.T) { + r := newLineageRig(t) + single := common.DeriveIdempotencyToken("lineage-97", 0) + + first, err := r.attempt("attempt-1") + require.NoError(t, err) + require.Equal(t, []string{single}, first, + "the first attempt of a plan with no accounts must keep the pre-#97 single-account token byte-for-byte") + + r.setPlanAccounts(shapeTestAccountA) + + retry, err := r.attempt("attempt-2") + require.NoError(t, err) + assert.Equal(t, []string{single}, retry, + "the retry must reuse the first attempt's token; a different token misses the provider's dedupe and double-buys (issue #97)") +} + +// TestRetryAfterPlanLosesLastAccountBuysNothing is the issue #97 regression +// guard, N-to-zero direction. +// +// The first attempt fanned out and bought acct-A under +// DeriveIdempotencyToken(K+":acct-A", 0). The account is then removed from the +// plan and the run is retried. The single-account path would derive +// DeriveIdempotencyToken(K, 0), which the provider has never seen, and there is +// no plan account left to fan out to, so the only safe outcome is to refuse and +// buy nothing. +func TestRetryAfterPlanLosesLastAccountBuysNothing(t *testing.T) { + r := newLineageRig(t) + r.setPlanAccounts(shapeTestAccountA) + + first, err := r.attempt("attempt-1") + require.NoError(t, err) + require.Equal(t, []string{common.DeriveIdempotencyToken("lineage-97:acct-A", 0)}, first, + "the first attempt of a plan with accounts must keep the pre-#97 fan-out token byte-for-byte") + + r.setPlanAccounts() + + retry, err := r.attempt("attempt-2") + require.Error(t, err, "the retry must be refused, not run on a path the first attempt did not take") + assert.Empty(t, retry, "nothing may be bought under a token the first attempt never used (issue #97)") +} + +// TestUnchangedPlanKeepsItsTokensAcrossRetries pins backward compatibility: a +// lineage whose plan never crosses zero-to-N accounts derives exactly the +// tokens it derived before issue #97, on every attempt and on either path. +func TestUnchangedPlanKeepsItsTokensAcrossRetries(t *testing.T) { + cases := []struct { + name string + accounts []config.CloudAccount + want string + }{ + {"no plan accounts: single-account path", nil, common.DeriveIdempotencyToken("lineage-97", 0)}, + {"plan accounts: fan-out path", []config.CloudAccount{shapeTestAccountA}, common.DeriveIdempotencyToken("lineage-97:acct-A", 0)}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + r := newLineageRig(t) + r.setPlanAccounts(tc.accounts...) + for _, id := range []string{"attempt-1", "attempt-2"} { + got, err := r.attempt(id) + require.NoError(t, err) + assert.Equal(t, []string{tc.want}, got, "%s must derive the pre-#97 token", id) + } + }) + } +} + +// TestFanOutShapePinFailureBuysNothing: when the shape cannot be pinned the +// executor cannot know which tokens a previous attempt used, so it must fail +// closed before any provider call. +func TestFanOutShapePinFailureBuysNothing(t *testing.T) { + r := newLineageRig(t) + r.setPlanAccounts(shapeTestAccountA) + r.store.PinExecutionFanOutShapeFn = func(_ context.Context, _, _ string) (string, error) { + return "", errors.New("db unavailable") + } + + got, err := r.attempt("attempt-1") + require.ErrorContains(t, err, "db unavailable") + assert.Empty(t, got, "nothing may be bought when the lineage's shape cannot be pinned") +} diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 1734a504..93174f41 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -368,6 +368,9 @@ func (m *mockConfigStoreForHealth) SaveExecutionIfStatus(ctx context.Context, e func (m *mockConfigStoreForHealth) SaveExecutionIfStatusTx(ctx context.Context, _ pgx.Tx, e *config.PurchaseExecution, _ []string) error { return m.SavePurchaseExecution(ctx, e) } +func (m *mockConfigStoreForHealth) PinExecutionFanOutShape(_ context.Context, _ string, shape string) (string, error) { + return shape, nil +} func (m *mockConfigStoreForHealth) RecoverReapedExecution(ctx context.Context, e *config.PurchaseExecution) error { return m.SavePurchaseExecution(ctx, e) } From 5dc755f0a5c4cb9cf45bdd98cd2c1294d7f9e9aa Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 00:21:19 +0100 Subject: [PATCH 2/3] fix(purchase): pin the fan-out shape only once an attempt can submit Address review of #896. An attempt refused before any provider call (e.g. the single-account path refusing recs that span accounts, or a credential failure) pinned its shape anyway, binding every retry to a path the operator may have reconfigured the plan to avoid. The path decision now only reads the pin. The pin is written right before submission: after resolveSingleAccountProvider on the single-account path, and after each account's credential resolution on the fan-out (always under the root lineage key). A concurrent attempt that pinned the other shape first makes this attempt refuse. The no-accounts refusal and the pin read/write failures on the single-account path are stamped beforeSubmission. Adds tests for the reviewer's scenario, colon keys never reaching the pin store, and the concurrent-pin refusal, and documents the lossy down migration. Co-Authored-By: Claude Opus 5.5 --- internal/analytics/collector_test.go | 3 + internal/config/interfaces.go | 3 + .../config/store_postgres_fan_out_shape.go | 18 ++ ...postgres_fan_out_shape_integration_test.go | 8 + ...purchase_execution_fan_out_shapes.down.sql | 6 + internal/mocks/fan_out_shape.go | 21 +++ internal/mocks/stores.go | 1 + internal/purchase/execution.go | 11 ++ internal/purchase/fan_out_shape.go | 123 +++++++++----- internal/purchase/fan_out_shape_test.go | 157 ++++++++++++++++-- internal/server/test_helpers_test.go | 3 + 11 files changed, 301 insertions(+), 53 deletions(-) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index cd51efac..c43b6a96 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -839,6 +839,9 @@ func (m *mockConfigStore) SaveExecutionIfStatus(ctx context.Context, e *config.P func (m *mockConfigStore) SaveExecutionIfStatusTx(ctx context.Context, _ pgx.Tx, e *config.PurchaseExecution, _ []string) error { return m.SavePurchaseExecution(ctx, e) } +func (m *mockConfigStore) GetExecutionFanOutShape(_ context.Context, _ string) (string, error) { + return "", nil +} func (m *mockConfigStore) PinExecutionFanOutShape(_ context.Context, _ string, shape string) (string, error) { return shape, nil } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 20484d57..6298b925 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -142,6 +142,9 @@ type StoreInterface interface { // FanOutShapeMulti) for the execution lineage lineageKey unless one is // already recorded, and returns the recorded shape (issue #97). PinExecutionFanOutShape(ctx context.Context, lineageKey, shape string) (string, error) + // GetExecutionFanOutShape returns the shape pinned for lineageKey, or "" + // when none is pinned yet (issue #97). + GetExecutionFanOutShape(ctx context.Context, lineageKey string) (string, error) GetPendingExecutions(ctx context.Context) ([]PurchaseExecution, error) // GetExecutionsByStatuses returns executions in any of the given states, // newest first, capped at `limit`. Used by the History handler to render diff --git a/internal/config/store_postgres_fan_out_shape.go b/internal/config/store_postgres_fan_out_shape.go index 5373dd19..7081b000 100644 --- a/internal/config/store_postgres_fan_out_shape.go +++ b/internal/config/store_postgres_fan_out_shape.go @@ -2,7 +2,10 @@ package config import ( "context" + "errors" "fmt" + + "github.com/jackc/pgx/v5" ) // Fan-out shapes of a purchase execution lineage (issue #97). The shape decides @@ -40,3 +43,18 @@ func (s *PostgresStore) PinExecutionFanOutShape(ctx context.Context, lineageKey, } return pinned, nil } + +// GetExecutionFanOutShape returns the fan-out shape pinned for the execution +// lineage keyed by lineageKey, or "" when none is pinned yet. +func (s *PostgresStore) GetExecutionFanOutShape(ctx context.Context, lineageKey string) (string, error) { + var shape string + err := s.db.QueryRow(ctx, + `SELECT shape FROM purchase_execution_fan_out_shapes WHERE idempotency_key = $1`, lineageKey).Scan(&shape) + if errors.Is(err, pgx.ErrNoRows) { + return "", nil + } + if err != nil { + return "", fmt.Errorf("failed to read fan-out shape for lineage %s: %w", lineageKey, err) + } + return shape, nil +} diff --git a/internal/config/store_postgres_fan_out_shape_integration_test.go b/internal/config/store_postgres_fan_out_shape_integration_test.go index 0ffa3612..dc2fe0b0 100644 --- a/internal/config/store_postgres_fan_out_shape_integration_test.go +++ b/internal/config/store_postgres_fan_out_shape_integration_test.go @@ -22,6 +22,10 @@ func TestPinExecutionFanOutShape(t *testing.T) { t.Run("first writer wins", func(t *testing.T) { key := uuid.NewString() + unpinned, err := store.GetExecutionFanOutShape(ctx, key) + require.NoError(t, err) + assert.Empty(t, unpinned, "a lineage nothing pinned reads as empty") + got, err := store.PinExecutionFanOutShape(ctx, key, FanOutShapeSingle) require.NoError(t, err) assert.Equal(t, FanOutShapeSingle, got) @@ -29,6 +33,10 @@ func TestPinExecutionFanOutShape(t *testing.T) { got, err = store.PinExecutionFanOutShape(ctx, key, FanOutShapeMulti) require.NoError(t, err) assert.Equal(t, FanOutShapeSingle, got, "a later attempt must get the first attempt's shape back") + + read, err := store.GetExecutionFanOutShape(ctx, key) + require.NoError(t, err) + assert.Equal(t, FanOutShapeSingle, read, "the read must return the pinned shape") }) t.Run("concurrent first attempts agree", func(t *testing.T) { diff --git a/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql index e1b97c8e..951f4bd2 100644 --- a/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql +++ b/internal/database/postgres/migrations/000123_purchase_execution_fan_out_shapes.down.sql @@ -1 +1,7 @@ +-- Reverse 000123: drop the fan-out shape pins (issue #97). +-- +-- Lossy by design: every pinned shape is dropped, and re-applying the up +-- migration restores only the "multi" pins its backfill can prove from +-- per-account rows. "single" pins are not recoverable, so their lineages fall +-- back to the live plan accounts until their next submission pins them again. DROP TABLE IF EXISTS purchase_execution_fan_out_shapes; diff --git a/internal/mocks/fan_out_shape.go b/internal/mocks/fan_out_shape.go index 1b494d84..5b66bf30 100644 --- a/internal/mocks/fan_out_shape.go +++ b/internal/mocks/fan_out_shape.go @@ -26,6 +26,12 @@ func (p *fanOutShapePins) pin(lineageKey, shape string) string { return shape } +func (p *fanOutShapePins) get(lineageKey string) string { + p.mu.Lock() + defer p.mu.Unlock() + return p.pins[lineageKey] +} + // PinExecutionFanOutShape mocks the fan-out shape pin (issue #97). // PinExecutionFanOutShapeFn takes priority; with no expectation registered it // behaves like the real store (first writer wins per lineage key), so one mock @@ -41,3 +47,18 @@ func (m *MockConfigStore) PinExecutionFanOutShape(ctx context.Context, lineageKe args := m.Called(ctx, lineageKey, shape) return args.String(0), args.Error(1) } + +// GetExecutionFanOutShape mocks the pinned-shape read (issue #97). +// GetExecutionFanOutShapeFn takes priority; with no expectation registered it +// reads what PinExecutionFanOutShape's default pinned on this mock. +func (m *MockConfigStore) GetExecutionFanOutShape(ctx context.Context, lineageKey string) (string, error) { + m.record("GetExecutionFanOutShape", ctx, lineageKey) + if m.GetExecutionFanOutShapeFn != nil { + return m.GetExecutionFanOutShapeFn(ctx, lineageKey) + } + if !isExpected(&m.Mock, "GetExecutionFanOutShape") { + return m.fanOutShapes.get(lineageKey), nil + } + args := m.Called(ctx, lineageKey) + return args.String(0), args.Error(1) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index ebb48a84..4ec7e0dd 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -53,6 +53,7 @@ type MockConfigStore struct { ListPendingExecutionIDsForAccountFn func(ctx context.Context, accountID string) ([]string, error) SavePurchaseExecutionFn func(ctx context.Context, exec *config.PurchaseExecution) error PinExecutionFanOutShapeFn func(ctx context.Context, lineageKey, shape string) (string, error) + GetExecutionFanOutShapeFn func(ctx context.Context, lineageKey string) (string, error) GetUserEmailByIDFn func(ctx context.Context, userID string) (string, error) mock.Mock // callLog records every invocation, including the ones served from a diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 60212f84..039159d2 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -107,6 +107,10 @@ func (m *Manager) executeSingleAccount(ctx context.Context, exec *config.Purchas if err != nil { return beforeSubmission(err) } + // Pinned only now that the attempt can submit (issue #97); see fan_out_shape.go. + if err := m.pinLineageShape(ctx, exec, config.FanOutShapeSingle); err != nil { + return err + } // #646: stamp the resolved target account on history, not the ambient // AWS host account. resolveSingleAccountProvider returns the target's // ExternalID (the provider-appropriate account identifier for AWS / @@ -290,6 +294,13 @@ func (m *Manager) executeForAccount(ctx context.Context, baseExec *config.Purcha // executeForAccount under the gocyclo:10 budget. return false, m.persistFailedAccountExecution(ctx, &acctExec, account, credErr) } + // Pinned on the ROOT lineage, and only once this account can submit (issue + // #97); see fan_out_shape.go. + if pinErr := m.pinLineageShape(ctx, baseExec, config.FanOutShapeMulti); pinErr != nil { + acctExec.Status = "failed" + acctExec.Error = pinErr.Error() + return false, m.persistFailedAccountExecution(ctx, &acctExec, account, pinErr) + } accountID := account.ExternalID totalSavings, totalUpfront, purchaseErrors, procErr := m.processPurchaseRecommendations(ctx, &acctExec, plan, accountID, provCfg) diff --git a/internal/purchase/fan_out_shape.go b/internal/purchase/fan_out_shape.go index be46c8ff..9a95a5f6 100644 --- a/internal/purchase/fan_out_shape.go +++ b/internal/purchase/fan_out_shape.go @@ -9,59 +9,104 @@ import ( "github.com/LeanerCloud/cloud-commitments-platform/internal/config" ) -// lineageFansOut decides whether a plan execution with no cloud_account_id fans -// out across the plan's accounts or runs the single-account path (issue #97). +// The single-account path and the per-account fan-out derive their provider +// idempotency tokens from different keys: the lineage key itself, and +// ":" per account. Choosing the path from the plan's +// CURRENT accounts let a plan that gained its first account, or lost its last +// one, between an attempt and its Retry derive tokens the provider had never +// seen, so a commitment the first attempt bought was bought again (issue #97). // -// The two paths derive their provider idempotency tokens from different keys: -// the lineage key itself on the single-account path, ":" on the fan-out path. Choosing the path from the plan's CURRENT accounts -// let a plan that gained its first account, or lost its last one, between an -// attempt and its Retry derive tokens the provider had never seen, so a -// commitment the first attempt bought was bought again. Instead, the first -// attempt of a lineage pins the shape it runs with, and every later attempt -// follows the pin: +// So the first attempt of a lineage that gets as far as submitting pins the +// shape it submitted under (pinLineageShape), and every later attempt follows +// the pin (lineageFansOut). An attempt refused before submission pins nothing: +// it derived no token that needs protecting, and pinning it would bind every +// retry to a path the operator may have just reconfigured the plan to avoid. + +// fanOutPinKey returns the lineage key a root plan execution pins its shape +// under, and false for rows that take no part in pinning: // -// - pinned shape matches the live one: unchanged behavior, so tokens of a -// lineage whose plan never crossed zero-to-N accounts are byte-identical. -// - pinned single, plan now has accounts: run the single-account path again, -// re-deriving the first attempt's tokens so the provider dedupes. -// - pinned multi, plan now has no accounts: refuse. The first attempt's tokens -// are per account and there is no account left to fan out to; the -// single-account path would derive tokens the provider has never seen. +// - direct purchases (no plan) and rows bound to an account: they always run +// the single-account path, whatever the plan's accounts are; +// - an empty key, which cannot identify a lineage; +// - a key with a colon: a per-account lineage (":") +// whose row reached the executor without its scope. reattachAccountScope +// (issue #1537) owns that case, and pinning it would write a per-account +// key into a table of root lineages. +func fanOutPinKey(exec *config.PurchaseExecution) (string, bool) { + if exec.PlanID == "" || exec.CloudAccountID != nil { + return "", false + } + key := idempotencyLineageKey(exec) + if key == "" || strings.Contains(key, ":") { + return "", false + } + return key, true +} + +// lineageFansOut decides whether a plan execution with no cloud_account_id fans +// out across the plan's accounts or runs the single-account path: // -// A key with a colon is a per-account lineage that reached the executor without -// its scope; reattachAccountScope (issue #1537) owns that case, so it is left to -// the live decision. An empty key cannot be pinned meaningfully and is left to -// the live decision too. +// - nothing pinned yet, or the pin matches the live shape: the live shape, so +// a lineage whose plan never crossed zero-to-N accounts derives exactly the +// tokens it derived before issue #97. +// - pinned single, plan now has accounts: the single-account path again, so +// the first submission's tokens are re-derived and the provider dedupes. +// - pinned multi, plan now has no accounts: refused. The submitted tokens are +// per account and there is no account left to fan out to; the +// single-account path would derive tokens the provider has never seen. func (m *Manager) lineageFansOut(ctx context.Context, exec *config.PurchaseExecution, accounts []config.CloudAccount) (bool, error) { live := len(accounts) > 0 - key := idempotencyLineageKey(exec) - if key == "" || strings.Contains(key, ":") { + key, ok := fanOutPinKey(exec) + if !ok { return live, nil } - want := config.FanOutShapeSingle - if live { - want = config.FanOutShapeMulti - } - pinned, err := m.config.PinExecutionFanOutShape(ctx, key, want) + pinned, err := m.config.GetExecutionFanOutShape(ctx, key) if err != nil { - return false, fmt.Errorf("execution %s: %w", exec.ExecutionID, err) + return false, beforeSubmission(fmt.Errorf("execution %s: %w", exec.ExecutionID, err)) } - switch pinned { - case want: + switch { + case pinned == "" || pinned == liveShape(live): return live, nil - case config.FanOutShapeSingle: - logging.Warnf("purchase[%s]: plan %s now has %d account(s), but this execution's lineage first ran on the "+ + case pinned == config.FanOutShapeSingle: + logging.Warnf("purchase[%s]: plan %s now has %d account(s), but this execution's lineage first submitted on the "+ "single-account path; re-driving that path so the provider idempotency tokens match (issue #97)", exec.ExecutionID, exec.PlanID, len(accounts)) return false, nil - case config.FanOutShapeMulti: - return false, fmt.Errorf("refusing execution %s: its lineage first fanned out across plan %s's accounts, "+ - "which now has none, so the per-account idempotency tokens cannot be re-derived and the single-account "+ - "path would buy again under tokens the provider has never seen (issue #97). "+ + case pinned == config.FanOutShapeMulti: + return false, beforeSubmission(fmt.Errorf("refusing execution %s: its lineage first submitted across plan %s's "+ + "accounts, which now has none, so the per-account idempotency tokens cannot be re-derived and the "+ + "single-account path would buy again under tokens the provider has never seen (issue #97). "+ "Re-attach the accounts to the plan to let the retry complete, or cancel this execution", - exec.ExecutionID, exec.PlanID) + exec.ExecutionID, exec.PlanID)) default: - return false, fmt.Errorf("execution %s: unknown pinned fan-out shape %q", exec.ExecutionID, pinned) + return false, beforeSubmission(fmt.Errorf("execution %s: unknown pinned fan-out shape %q", exec.ExecutionID, pinned)) + } +} + +// pinLineageShape pins shape for exec's lineage. Callers invoke it after the +// attempt's last lineage-level pre-submission check and before any provider +// call. A pin already holding the other shape means a concurrent attempt of the +// same lineage submitted under it first, so this attempt must not submit. +func (m *Manager) pinLineageShape(ctx context.Context, exec *config.PurchaseExecution, shape string) error { + key, ok := fanOutPinKey(exec) + if !ok { + return nil + } + pinned, err := m.config.PinExecutionFanOutShape(ctx, key, shape) + if err != nil { + return beforeSubmission(fmt.Errorf("execution %s: %w", exec.ExecutionID, err)) + } + if pinned != shape { + return beforeSubmission(fmt.Errorf("refusing execution %s: another attempt of its lineage submitted on the %s "+ + "fan-out shape first, and submitting on the %s shape would derive idempotency tokens the provider has "+ + "never seen (issue #97); retry it", exec.ExecutionID, pinned, shape)) + } + return nil +} + +func liveShape(fansOut bool) string { + if fansOut { + return config.FanOutShapeMulti } + return config.FanOutShapeSingle } diff --git a/internal/purchase/fan_out_shape_test.go b/internal/purchase/fan_out_shape_test.go index 91f51778..e1f66db6 100644 --- a/internal/purchase/fan_out_shape_test.go +++ b/internal/purchase/fan_out_shape_test.go @@ -30,9 +30,14 @@ type lineageRig struct { // shapeTestAccountA is a cloud account that exists in config throughout; only // its membership in the plan changes between attempts. -var shapeTestAccountA = config.CloudAccount{ - ID: "acct-A", Name: "A", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys", -} +var ( + shapeTestAccountA = config.CloudAccount{ + ID: "acct-A", Name: "A", Provider: "aws", ExternalID: "111111111111", AWSAuthMode: "access_keys", + } + shapeTestAccountB = config.CloudAccount{ + ID: "acct-B", Name: "B", Provider: "aws", ExternalID: "222222222222", AWSAuthMode: "access_keys", + } +) func newLineageRig(t *testing.T) *lineageRig { t.Helper() @@ -51,9 +56,10 @@ func newLineageRig(t *testing.T) *lineageRig { return append([]config.CloudAccount(nil), r.planAccounts...), nil } r.store.GetCloudAccountFn = func(_ context.Context, id string) (*config.CloudAccount, error) { - if id == shapeTestAccountA.ID { - acct := shapeTestAccountA - return &acct, nil + for _, acct := range []config.CloudAccount{shapeTestAccountA, shapeTestAccountB} { + if acct.ID == id { + return &acct, nil + } } return nil, nil } @@ -94,17 +100,23 @@ func (r *lineageRig) setPlanAccounts(accounts ...config.CloudAccount) { // Each attempt is a fresh row, as a Retry successor is: a new ExecutionID, the // same IdempotencyKey, no cloud_account_id and unpurchased recommendations. func (r *lineageRig) attempt(executionID string) ([]string, error) { + r.t.Helper() + return r.attemptKeyed(executionID, "lineage-97", scopedTestRec("acct-A")) +} + +// attemptKeyed is attempt with an explicit lineage key and recommendations. +func (r *lineageRig) attemptKeyed(executionID, lineageKey string, recs ...config.RecommendationRecord) ([]string, error) { r.t.Helper() r.mu.Lock() r.tokens = nil r.mu.Unlock() exec := &config.PurchaseExecution{ ExecutionID: executionID, - IdempotencyKey: "lineage-97", + IdempotencyKey: lineageKey, PlanID: "plan-x", StepNumber: 1, Source: common.PurchaseSourceWeb, - Recommendations: []config.RecommendationRecord{scopedTestRec("acct-A")}, + Recommendations: recs, } err := r.manager.executePurchase(context.Background(), exec) r.mu.Lock() @@ -164,6 +176,8 @@ func TestRetryAfterPlanLosesLastAccountBuysNothing(t *testing.T) { retry, err := r.attempt("attempt-2") require.Error(t, err, "the retry must be refused, not run on a path the first attempt did not take") assert.Empty(t, retry, "nothing may be bought under a token the first attempt never used (issue #97)") + var notSubmitted *beforeSubmissionError + assert.ErrorAs(t, err, ¬Submitted, "the refusal happens before any provider call and must say so") } // TestUnchangedPlanKeepsItsTokensAcrossRetries pins backward compatibility: a @@ -192,16 +206,131 @@ func TestUnchangedPlanKeepsItsTokensAcrossRetries(t *testing.T) { } // TestFanOutShapePinFailureBuysNothing: when the shape cannot be pinned the -// executor cannot know which tokens a previous attempt used, so it must fail -// closed before any provider call. +// executor cannot know which tokens another attempt used, so it must fail +// closed before any provider call. On the single-account path the refusal is +// stamped as before-submission; on the fan-out it fails the account like a +// credential failure does. func TestFanOutShapePinFailureBuysNothing(t *testing.T) { + for _, tc := range []struct { + name string + accounts []config.CloudAccount + }{ + {"single-account path", nil}, + {"fan-out path", []config.CloudAccount{shapeTestAccountA}}, + } { + t.Run(tc.name, func(t *testing.T) { + r := newLineageRig(t) + r.setPlanAccounts(tc.accounts...) + r.store.PinExecutionFanOutShapeFn = func(_ context.Context, _, _ string) (string, error) { + return "", errors.New("db unavailable") + } + + got, err := r.attempt("attempt-1") + require.ErrorContains(t, err, "db unavailable") + assert.Empty(t, got, "nothing may be bought when the lineage's shape cannot be pinned") + if tc.accounts == nil { + var notSubmitted *beforeSubmissionError + assert.ErrorAs(t, err, ¬Submitted, "the refusal happens before any provider call and must say so") + } + }) + } +} + +// TestRetryAfterPreSubmissionRefusalFollowsReconfiguredPlan: an attempt refused +// before any provider call derived no token that needs protecting, so it must +// not pin its shape. Here the plan has no accounts and the step's recs span two +// accounts, which the single-account path refuses (#1902). The operator then +// attaches both accounts so the step can fan out; the retry must fan out and +// buy each account once, not be bound to the single-account path and refused +// on every retry. +func TestRetryAfterPreSubmissionRefusalFollowsReconfiguredPlan(t *testing.T) { + r := newLineageRig(t) + recs := func() []config.RecommendationRecord { + b := scopedTestRec("acct-B") + b.ResourceType = "c5.large" + return []config.RecommendationRecord{scopedTestRec("acct-A"), b} + } + + first, err := r.attemptKeyed("attempt-1", "lineage-97", recs()...) + require.Error(t, err, "the single-account path must refuse recs that span accounts") + require.Empty(t, first) + pinned, err := r.store.GetExecutionFanOutShape(context.Background(), "lineage-97") + require.NoError(t, err) + assert.Empty(t, pinned, "an attempt that never reached submission must not pin the lineage's shape") + + r.setPlanAccounts(shapeTestAccountA, shapeTestAccountB) + + retry, err := r.attemptKeyed("attempt-2", "lineage-97", recs()...) + require.NoError(t, err) + want := []string{ + common.DeriveIdempotencyToken("lineage-97:acct-A", 0), + common.DeriveIdempotencyToken("lineage-97:acct-B", 0), + } + sort.Strings(want) + assert.Equal(t, want, retry, "the retry must fan out across the newly attached accounts") +} + +// TestColonLineageKeysAreNeverPinned: only root lineages are pinned. A +// scopeless per-account row (":", issue #1537) must +// neither read nor write a pin under its own key, whichever path it takes, and +// a fan-out must pin its ROOT key, never a per-account one. +func TestColonLineageKeysAreNeverPinned(t *testing.T) { + for _, tc := range []struct { + name string + accounts []config.CloudAccount + }{ + {"account still on the plan: re-attached single-account re-drive", []config.CloudAccount{shapeTestAccountA}}, + {"plan has no accounts: single-account path", nil}, + } { + t.Run("scopeless per-account row, "+tc.name, func(t *testing.T) { + r := newLineageRig(t) + r.setPlanAccounts(tc.accounts...) + + got, err := r.attemptKeyed("successor", "root-lineage:acct-A", scopedTestRec("acct-A")) + require.NoError(t, err) + require.Equal(t, []string{common.DeriveIdempotencyToken("root-lineage:acct-A", 0)}, got) + store := r.store + store.AssertNotCalled(t, "GetExecutionFanOutShape") + store.AssertNotCalled(t, "PinExecutionFanOutShape") + }) + } + + t.Run("root fan-out pins the root key", func(t *testing.T) { + r := newLineageRig(t) + r.setPlanAccounts(shapeTestAccountA) + var mu sync.Mutex + var keys []string + r.store.PinExecutionFanOutShapeFn = func(_ context.Context, key, shape string) (string, error) { + mu.Lock() + keys = append(keys, key) + mu.Unlock() + return shape, nil + } + + _, err := r.attempt("attempt-1") + require.NoError(t, err) + mu.Lock() + defer mu.Unlock() + require.NotEmpty(t, keys, "a fan-out that submits must pin its lineage") + for _, k := range keys { + assert.Equal(t, "lineage-97", k, "only the root lineage key may be pinned") + } + }) +} + +// TestConcurrentAttemptPinnedOtherShapeBuysNothing: the read said nothing was +// pinned, but by the time this attempt pins, a concurrent attempt of the same +// lineage pinned the other shape and submitted under it. This attempt must not +// submit under its own shape's tokens. +func TestConcurrentAttemptPinnedOtherShapeBuysNothing(t *testing.T) { r := newLineageRig(t) - r.setPlanAccounts(shapeTestAccountA) r.store.PinExecutionFanOutShapeFn = func(_ context.Context, _, _ string) (string, error) { - return "", errors.New("db unavailable") + return config.FanOutShapeMulti, nil } got, err := r.attempt("attempt-1") - require.ErrorContains(t, err, "db unavailable") - assert.Empty(t, got, "nothing may be bought when the lineage's shape cannot be pinned") + require.Error(t, err) + assert.Empty(t, got, "nothing may be bought under the losing shape's tokens") + var notSubmitted *beforeSubmissionError + assert.ErrorAs(t, err, ¬Submitted) } diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 93174f41..4cfdb7c4 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -368,6 +368,9 @@ func (m *mockConfigStoreForHealth) SaveExecutionIfStatus(ctx context.Context, e func (m *mockConfigStoreForHealth) SaveExecutionIfStatusTx(ctx context.Context, _ pgx.Tx, e *config.PurchaseExecution, _ []string) error { return m.SavePurchaseExecution(ctx, e) } +func (m *mockConfigStoreForHealth) GetExecutionFanOutShape(_ context.Context, _ string) (string, error) { + return "", nil +} func (m *mockConfigStoreForHealth) PinExecutionFanOutShape(_ context.Context, _ string, shape string) (string, error) { return shape, nil } From dec4b318770dda4ee14fbd9f56eb605e3bf64d42 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Fri, 16 Oct 2026 01:05:22 +0100 Subject: [PATCH 3/3] refactor(purchase): keep executeSingleAccount under the gocyclo budget Extract the single-account pre-submission steps (row-account check, provider resolution, fan-out shape pin) into prepareSingleAccountSubmission. No behavior change. Co-Authored-By: Claude Opus 5.5 --- internal/purchase/execution.go | 29 +++++++++++++++++++++-------- 1 file changed, 21 insertions(+), 8 deletions(-) diff --git a/internal/purchase/execution.go b/internal/purchase/execution.go index 039159d2..ebf6a6d6 100644 --- a/internal/purchase/execution.go +++ b/internal/purchase/execution.go @@ -100,15 +100,8 @@ func (m *Manager) executePurchase(ctx context.Context, exec *config.PurchaseExec // plain error when nothing committed. Split out of executePurchase to keep that // function under the gocyclo budget. func (m *Manager) executeSingleAccount(ctx context.Context, exec *config.PurchaseExecution, plan *config.PurchasePlan) error { - if err := requireRecsMatchRowAccount(exec); err != nil { - return beforeSubmission(err) - } - provCfg, targetAccountID, err := m.resolveSingleAccountProvider(ctx, exec) + provCfg, targetAccountID, err := m.prepareSingleAccountSubmission(ctx, exec) if err != nil { - return beforeSubmission(err) - } - // Pinned only now that the attempt can submit (issue #97); see fan_out_shape.go. - if err := m.pinLineageShape(ctx, exec, config.FanOutShapeSingle); err != nil { return err } // #646: stamp the resolved target account on history, not the ambient @@ -158,6 +151,26 @@ func (m *Manager) executeSingleAccount(ctx context.Context, exec *config.Purchas return nil } +// prepareSingleAccountSubmission runs the single-account path's pre-submission +// steps: check the recs match the row's account, resolve the provider and +// target account, then pin the lineage's shape now that the attempt can submit +// (issue #97; see fan_out_shape.go). Every error it returns is before +// submission. Split out of executeSingleAccount to keep that function under +// the gocyclo budget. +func (m *Manager) prepareSingleAccountSubmission(ctx context.Context, exec *config.PurchaseExecution) (*provider.ProviderConfig, string, error) { + if err := requireRecsMatchRowAccount(exec); err != nil { + return nil, "", beforeSubmission(err) + } + provCfg, targetAccountID, err := m.resolveSingleAccountProvider(ctx, exec) + if err != nil { + return nil, "", beforeSubmission(err) + } + if err := m.pinLineageShape(ctx, exec, config.FanOutShapeSingle); err != nil { + return nil, "", err // already beforeSubmission + } + return provCfg, targetAccountID, nil +} + // anyRecPurchased reports whether at least one recommendation in the slice // was marked Purchased=true by the aggregator (a real commitment was created // on the cloud provider). It is the gate that distinguishes a partial success