From 6e2fff936f7b8332af88d9bd99d6ea3491bceacd Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 16 Jul 2026 11:55:19 +0300 Subject: [PATCH 1/2] feat(exchange): scope pending-cancellation, LadderRunID plumbing, true dry-run (G10 / #1348) RunAutoExchange previously called CancelAllPendingExchanges (store-wide), so the standalone ri_exchange_reshape task would silently cancel pending reshapes created by ladder runs and vice versa. Three sub-goals addressed: 1. SCOPE the cancellation: replace CancelAllPendingExchanges in RunAutoExchange with CancelPendingExchangesByOrigin(ctx, ladderScoped). - ladderScoped=false (standalone): WHERE ladder_run_id IS NULL - ladderScoped=true (ladder): WHERE ladder_run_id IS NOT NULL Keep CancelAllPendingExchanges on the interface for compat. 2. LADDER_RUN_ID plumbing: add LadderRunID *string to ExchangeRecord, RunAutoExchangeParams, and config.RIExchangeRecord; write it in SaveRIExchangeRecord ($21 param) and read it in all SELECT queries / queryRIExchangeRecords Scan; propagate through the adapter converters. The DB column already existed (migration 000080). 3. TRUE DRY-RUN: RunAutoExchangeParams.DryRun=true skips cancellation, SaveRIExchangeRecord, and Execute; returns outcomes tagged Simulated=true. No live approval tokens issued. ReshapeBuffer's DryRun rejection comment updated to reflect that the pkg/exchange layer now supports it (end-to-end wiring deferred to L16). Also: extract nullPtrFromNullString helper in store_postgres.go to reduce queryRIExchangeRecords cyclomatic complexity back below 10; add ExchangeOrigin typed enum (ExchangeOriginStandalone / ExchangeOriginLadder). Tests added: - TestRunAutoExchange_StandaloneDoesNotCancelLadderPendings (regression for G10) - TestRunAutoExchange_LadderDoesNotCancelStandalonePendings - TestRunAutoExchange_DryRun_ManualMode_ZeroMutations (mock.AssertExpectations) - TestRunAutoExchange_DryRun_AutoMode_ZeroMutations - TestRunAutoExchange_LadderRunID_StampedOnRecords (round-trip) - TestRunAutoExchange_NoLadderRunID_RecordHasNilLadderRunID - TestPGXMock_CancelPendingExchangesByOrigin_{Standalone,Ladder} - TestPostgresStore_CancelPendingExchangesByOrigin_NilDB --- internal/analytics/collector_test.go | 4 + internal/config/interfaces.go | 7 + internal/config/store_postgres.go | 71 ++++-- internal/config/store_postgres_nil_db_test.go | 12 ++ .../config/store_postgres_pgxmock_test.go | 32 ++- internal/config/types.go | 8 +- internal/mocks/stores.go | 9 + internal/server/handler_ri_exchange.go | 6 + internal/server/handler_ri_exchange_test.go | 20 +- internal/server/test_helpers_test.go | 4 + pkg/exchange/auto.go | 102 +++++++-- pkg/exchange/auto_test.go | 202 ++++++++++++++++++ pkg/go.mod | 1 + pkg/go.sum | 2 + providers/aws/ladder/reshape.go | 25 ++- 15 files changed, 457 insertions(+), 48 deletions(-) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 70e7359e8..ed989b6c2 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -320,6 +320,10 @@ func (m *mockConfigStore) GetRIExchangeDailySpend(ctx context.Context, date time func (m *mockConfigStore) CancelAllPendingExchanges(ctx context.Context) (int64, error) { return 0, nil } + +func (m *mockConfigStore) CancelPendingExchangesByOrigin(_ context.Context, _ bool) (int64, error) { + return 0, nil +} func (m *mockConfigStore) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]config.RIExchangeRecord, error) { return nil, nil } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 5af9a9942..b75a98a5b 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -209,6 +209,13 @@ type StoreInterface interface { FailRIExchange(ctx context.Context, id string, errorMsg string) error GetRIExchangeDailySpend(ctx context.Context, date time.Time) (string, error) CancelAllPendingExchanges(ctx context.Context) (int64, error) + // CancelPendingExchangesByOrigin cancels only pending records that match the + // given origin scope: + // - ladderScoped=false (standalone): cancels WHERE ladder_run_id IS NULL + // - ladderScoped=true (ladder): cancels WHERE ladder_run_id IS NOT NULL + // This prevents the standalone ri_exchange_reshape task from wiping out + // ladder-linked pending reshapes and vice versa (gap G10 / issue #1348). + CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]RIExchangeRecord, error) // Cloud accounts diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 81694fec6..3b7754fc0 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -2200,8 +2200,8 @@ func (s *PostgresStore) SaveRIExchangeRecord(ctx context.Context, record *RIExch source_instance_type, source_count, target_offering_id, target_instance_type, target_count, payment_due, status, approval_token, error, mode, completed_at, expires_at, - created_at, updated_at, created_by_user_id - ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20) + created_at, updated_at, created_by_user_id, ladder_run_id + ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17, $18, $19, $20, $21) ` _, err := s.db.Exec(ctx, query, @@ -2225,6 +2225,7 @@ func (s *PostgresStore) SaveRIExchangeRecord(ctx context.Context, record *RIExch record.CreatedAt, record.UpdatedAt, record.CreatedByUserID, + record.LadderRunID, ) if err != nil { @@ -2242,7 +2243,7 @@ func (s *PostgresStore) GetRIExchangeRecord(ctx context.Context, id string) (*RI target_instance_type, target_count, payment_due::text, status, approval_token, error, mode, created_at, updated_at, completed_at, expires_at, - created_by_user_id, approved_by + created_by_user_id, approved_by, ladder_run_id FROM ri_exchange_history WHERE id = $1 ` @@ -2267,7 +2268,7 @@ func (s *PostgresStore) GetRIExchangeRecordByToken(ctx context.Context, token st target_instance_type, target_count, payment_due::text, status, approval_token, error, mode, created_at, updated_at, completed_at, expires_at, - created_by_user_id, approved_by + created_by_user_id, approved_by, ladder_run_id FROM ri_exchange_history WHERE approval_token = $1 ` @@ -2292,7 +2293,7 @@ func (s *PostgresStore) GetRIExchangeHistory(ctx context.Context, since time.Tim target_instance_type, target_count, payment_due::text, status, approval_token, error, mode, created_at, updated_at, completed_at, expires_at, - created_by_user_id, approved_by + created_by_user_id, approved_by, ladder_run_id FROM ri_exchange_history WHERE created_at >= $1 ORDER BY created_at DESC @@ -2317,7 +2318,7 @@ func (s *PostgresStore) TransitionRIExchangeStatus(ctx context.Context, id strin target_instance_type, target_count, payment_due::text, status, approval_token, error, mode, created_at, updated_at, completed_at, expires_at, - created_by_user_id, approved_by + created_by_user_id, approved_by, ladder_run_id ` records, err := s.queryRIExchangeRecords(ctx, query, id, fromStatus, toStatus, actor) @@ -2432,7 +2433,9 @@ func (s *PostgresStore) GetRIExchangeDailySpend(ctx context.Context, date time.T return total, nil } -// CancelAllPendingExchanges cancels all pending RI exchange records. +// CancelAllPendingExchanges cancels all pending RI exchange records regardless +// of origin. Kept for interface compatibility; new callers should prefer +// CancelPendingExchangesByOrigin to avoid cross-origin contamination. func (s *PostgresStore) CancelAllPendingExchanges(ctx context.Context) (int64, error) { query := ` UPDATE ri_exchange_history @@ -2448,6 +2451,36 @@ func (s *PostgresStore) CancelAllPendingExchanges(ctx context.Context) (int64, e return result.RowsAffected(), nil } +// CancelPendingExchangesByOrigin cancels only pending records that match the +// given origin scope (gap G10 / issue #1348): +// - ladderScoped=false (standalone): cancels WHERE ladder_run_id IS NULL +// - ladderScoped=true (ladder): cancels WHERE ladder_run_id IS NOT NULL +func (s *PostgresStore) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { + var query string + if ladderScoped { + query = ` + UPDATE ri_exchange_history + SET status = 'cancelled' + WHERE status = 'pending' + AND ladder_run_id IS NOT NULL + ` + } else { + query = ` + UPDATE ri_exchange_history + SET status = 'cancelled' + WHERE status = 'pending' + AND ladder_run_id IS NULL + ` + } + + result, err := s.db.Exec(ctx, query) + if err != nil { + return 0, fmt.Errorf("failed to cancel pending exchanges by origin (ladder_scoped=%v): %w", ladderScoped, err) + } + + return result.RowsAffected(), nil +} + // GetStaleProcessingExchanges returns processing exchanges older than the given duration. func (s *PostgresStore) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]RIExchangeRecord, error) { query := ` @@ -2456,7 +2489,7 @@ func (s *PostgresStore) GetStaleProcessingExchanges(ctx context.Context, olderTh target_instance_type, target_count, payment_due::text, status, approval_token, error, mode, created_at, updated_at, completed_at, expires_at, - created_by_user_id, approved_by + created_by_user_id, approved_by, ladder_run_id FROM ri_exchange_history WHERE status = 'processing' AND updated_at < NOW() - $1::interval ` @@ -2477,7 +2510,7 @@ func (s *PostgresStore) queryRIExchangeRecords(ctx context.Context, query string var record RIExchangeRecord var approvalToken, errStr sql.NullString var completedAt, expiresAt sql.NullTime - var createdByUserID, approvedBy sql.NullString + var createdByUserID, approvedBy, ladderRunID sql.NullString err := rows.Scan( &record.ID, @@ -2501,6 +2534,7 @@ func (s *PostgresStore) queryRIExchangeRecords(ctx context.Context, query string &expiresAt, &createdByUserID, &approvedBy, + &ladderRunID, ) if err != nil { return nil, fmt.Errorf("failed to scan ri exchange record: %w", err) @@ -2518,12 +2552,9 @@ func (s *PostgresStore) queryRIExchangeRecords(ctx context.Context, query string if expiresAt.Valid { record.ExpiresAt = &expiresAt.Time } - if createdByUserID.Valid { - record.CreatedByUserID = &createdByUserID.String - } - if approvedBy.Valid { - record.ApprovedBy = &approvedBy.String - } + record.CreatedByUserID = nullPtrFromNullString(createdByUserID) + record.ApprovedBy = nullPtrFromNullString(approvedBy) + record.LadderRunID = nullPtrFromNullString(ladderRunID) records = append(records, record) } @@ -3243,3 +3274,13 @@ func nullStringFromString(s string) sql.NullString { } return sql.NullString{String: s, Valid: true} } + +// nullPtrFromNullString converts a scanned sql.NullString to a *string for +// optional pointer fields. Returns nil when the DB value is NULL. +func nullPtrFromNullString(ns sql.NullString) *string { + if !ns.Valid { + return nil + } + s := ns.String + return &s +} diff --git a/internal/config/store_postgres_nil_db_test.go b/internal/config/store_postgres_nil_db_test.go index 152c05919..e52247c4e 100644 --- a/internal/config/store_postgres_nil_db_test.go +++ b/internal/config/store_postgres_nil_db_test.go @@ -165,6 +165,18 @@ func TestPostgresStore_CancelAllPendingExchanges_NilDB(t *testing.T) { assert.True(t, panicked, "expected panic with nil db connection") } +// TestPostgresStore_CancelPendingExchangesByOrigin_NilDB exercises the method entry. +func TestPostgresStore_CancelPendingExchangesByOrigin_NilDB(t *testing.T) { + store := NewPostgresStore(nil) + ctx := context.Background() + + panicked := callWithRecover(func() { + _, _ = store.CancelPendingExchangesByOrigin(ctx, false) + }) + + assert.True(t, panicked, "expected panic with nil db connection") +} + // TestPostgresStore_GetStaleProcessingExchanges_NilDB exercises the method entry. func TestPostgresStore_GetStaleProcessingExchanges_NilDB(t *testing.T) { store := NewPostgresStore(nil) diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index 3f6482235..795319015 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -988,6 +988,7 @@ func riExchangeRow(now time.Time) []interface{} { "manual", now, now, sql.NullTime{}, sql.NullTime{}, sql.NullString{}, sql.NullString{}, // created_by_user_id, approved_by + sql.NullString{}, // ladder_run_id (NULL for standalone) } } @@ -997,7 +998,7 @@ var riExchangeCols = []string{ "target_instance_type", "target_count", "payment_due", "status", "approval_token", "error", "mode", "created_at", "updated_at", "completed_at", "expires_at", - "created_by_user_id", "approved_by", + "created_by_user_id", "approved_by", "ladder_run_id", } func TestPGXMock_GetRIExchangeRecord_Success(t *testing.T) { @@ -1032,6 +1033,7 @@ func TestPGXMock_GetRIExchangeRecord_WithTimestamps(t *testing.T) { "auto", now, now, sql.NullTime{Valid: true, Time: now}, sql.NullTime{Valid: true, Time: now}, sql.NullString{}, sql.NullString{}, // created_by_user_id, approved_by + sql.NullString{}, // ladder_run_id } rows := pgxmock.NewRows(riExchangeCols).AddRow(row...) mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows) @@ -1815,6 +1817,34 @@ func TestPGXMock_CancelAllPendingExchanges_Success(t *testing.T) { assert.Equal(t, int64(3), n) } +// ─── CancelPendingExchangesByOrigin ────────────────────────────────────────── + +func TestPGXMock_CancelPendingExchangesByOrigin_Standalone(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + ctx := context.Background() + + // ladderScoped=false -> WHERE ladder_run_id IS NULL + mock.ExpectExec("UPDATE").WillReturnResult(pgxmock.NewResult("UPDATE", 2)) + + n, err := store.CancelPendingExchangesByOrigin(ctx, false) + require.NoError(t, err) + assert.Equal(t, int64(2), n) +} + +func TestPGXMock_CancelPendingExchangesByOrigin_Ladder(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + ctx := context.Background() + + // ladderScoped=true -> WHERE ladder_run_id IS NOT NULL + mock.ExpectExec("UPDATE").WillReturnResult(pgxmock.NewResult("UPDATE", 1)) + + n, err := store.CancelPendingExchangesByOrigin(ctx, true) + require.NoError(t, err) + assert.Equal(t, int64(1), n) +} + // ─── CleanupOldExecutions ───────────────────────────────────────────────────── func TestPGXMock_CleanupOldExecutions_Success(t *testing.T) { diff --git a/internal/config/types.go b/internal/config/types.go index 90f252877..f0b93cb74 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -804,7 +804,13 @@ type RIExchangeRecord struct { CreatedByUserID *string `json:"created_by_user_id,omitempty"` // ApprovedBy carries the email of the session user who approved the exchange // via the dashboard Approve button (issue #300). Nil for token-authed approvals. - ApprovedBy *string `json:"approved_by,omitempty"` + ApprovedBy *string `json:"approved_by,omitempty"` + // LadderRunID links this exchange record to the ladder run that created it + // (cudly-ladder engine). Nil for standalone ri_exchange_reshape task records. + // The database column ri_exchange_history.ladder_run_id was added in migration + // 000080 and is the authoritative source for origin scoping in + // CancelPendingExchangesByOrigin. + LadderRunID *string `json:"ladder_run_id,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` CompletedAt *time.Time `json:"completed_at,omitempty"` diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index f1bcbfdd3..fa263fb58 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -526,6 +526,15 @@ func (m *MockConfigStore) CancelAllPendingExchanges(ctx context.Context) (int64, return v, args.Error(1) } +func (m *MockConfigStore) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { + args := m.Called(ctx, ladderScoped) + v, ok := args.Get(0).(int64) + if !ok { + panic(fmt.Sprintf("mock: expected int64, got %T", args.Get(0))) + } + return v, args.Error(1) +} + func (m *MockConfigStore) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]config.RIExchangeRecord, error) { args := m.Called(ctx, olderThan) if args.Get(0) == nil { diff --git a/internal/server/handler_ri_exchange.go b/internal/server/handler_ri_exchange.go index e687d2505..fb9713781 100644 --- a/internal/server/handler_ri_exchange.go +++ b/internal/server/handler_ri_exchange.go @@ -294,6 +294,10 @@ func (a *configExchangeStoreAdapter) CancelAllPendingExchanges(ctx context.Conte return a.store.CancelAllPendingExchanges(ctx) } +func (a *configExchangeStoreAdapter) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { + return a.store.CancelPendingExchangesByOrigin(ctx, ladderScoped) +} + func (a *configExchangeStoreAdapter) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]exchange.ExchangeRecord, error) { cfgRecords, err := a.store.GetStaleProcessingExchanges(ctx, olderThan) if err != nil { @@ -335,6 +339,7 @@ func exchangeToConfigRecord(r *exchange.ExchangeRecord) *config.RIExchangeRecord ApprovalToken: r.ApprovalToken, Error: r.Error, Mode: r.Mode, + LadderRunID: r.LadderRunID, CreatedAt: r.CreatedAt, UpdatedAt: r.UpdatedAt, CompletedAt: r.CompletedAt, @@ -359,6 +364,7 @@ func configToExchangeRecord(r *config.RIExchangeRecord) exchange.ExchangeRecord ApprovalToken: r.ApprovalToken, Error: r.Error, Mode: r.Mode, + LadderRunID: r.LadderRunID, CreatedAt: r.CreatedAt, UpdatedAt: r.UpdatedAt, CompletedAt: r.CompletedAt, diff --git a/internal/server/handler_ri_exchange_test.go b/internal/server/handler_ri_exchange_test.go index 0fe391f93..fe7106577 100644 --- a/internal/server/handler_ri_exchange_test.go +++ b/internal/server/handler_ri_exchange_test.go @@ -790,12 +790,13 @@ func TestExecuteRIExchangeReshape_DailyCapHitMidRun(t *testing.T) { // --- Mock types --- type mockConfigStoreForExchange struct { - mockConfigStoreForHealth // embed base mock for unused methods - globalConfig *config.GlobalConfig - globalConfigErr error - saveRIExchangeRecordFunc func(ctx context.Context, record *config.RIExchangeRecord) error - cancelAllPendingFunc func(ctx context.Context) (int64, error) - getDailySpendFunc func(ctx context.Context, date time.Time) (string, error) + mockConfigStoreForHealth // embed base mock for unused methods + globalConfig *config.GlobalConfig + globalConfigErr error + saveRIExchangeRecordFunc func(ctx context.Context, record *config.RIExchangeRecord) error + cancelAllPendingFunc func(ctx context.Context) (int64, error) + cancelPendingByOriginFunc func(ctx context.Context, ladderScoped bool) (int64, error) + getDailySpendFunc func(ctx context.Context, date time.Time) (string, error) } func (m *mockConfigStoreForExchange) GetGlobalConfig(ctx context.Context) (*config.GlobalConfig, error) { @@ -822,6 +823,13 @@ func (m *mockConfigStoreForExchange) CancelAllPendingExchanges(ctx context.Conte return 0, nil } +func (m *mockConfigStoreForExchange) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { + if m.cancelPendingByOriginFunc != nil { + return m.cancelPendingByOriginFunc(ctx, ladderScoped) + } + return 0, nil +} + func (m *mockConfigStoreForExchange) GetRIExchangeDailySpend(ctx context.Context, date time.Time) (string, error) { if m.getDailySpendFunc != nil { return m.getDailySpendFunc(ctx, date) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index e37552e89..e0e2901e0 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -170,6 +170,10 @@ func (m *mockConfigStoreForHealth) GetRIExchangeDailySpend(ctx context.Context, func (m *mockConfigStoreForHealth) CancelAllPendingExchanges(ctx context.Context) (int64, error) { return 0, nil } + +func (m *mockConfigStoreForHealth) CancelPendingExchangesByOrigin(_ context.Context, _ bool) (int64, error) { + return 0, nil +} func (m *mockConfigStoreForHealth) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]config.RIExchangeRecord, error) { return nil, nil } diff --git a/pkg/exchange/auto.go b/pkg/exchange/auto.go index 76238b3b0..41c9e38b5 100644 --- a/pkg/exchange/auto.go +++ b/pkg/exchange/auto.go @@ -10,6 +10,20 @@ import ( "github.com/LeanerCloud/CUDly/pkg/logging" ) +// ExchangeOrigin identifies the originating task that created an exchange record, +// enabling scoped cancellation so standalone and ladder-originated pending +// exchanges do not contaminate each other. +type ExchangeOrigin string + +const ( + // ExchangeOriginStandalone is the standalone ri_exchange_reshape scheduled task. + // Its pending records have ladder_run_id IS NULL in the database. + ExchangeOriginStandalone ExchangeOrigin = "standalone-ri-exchange" + // ExchangeOriginLadder is the cudly-ladder engine (ladder run reshape phase). + // Its pending records have ladder_run_id IS NOT NULL in the database. + ExchangeOriginLadder ExchangeOrigin = "cudly-ladder" +) + // ExchangeRecord is a lightweight record type for the auto exchange logic. // It mirrors config.RIExchangeRecord but lives in pkg/exchange to avoid // cross-module imports (pkg/ is a separate Go module from internal/). @@ -29,16 +43,31 @@ type ExchangeRecord struct { ApprovalToken string Error string Mode string - CreatedAt time.Time - UpdatedAt time.Time - CompletedAt *time.Time - ExpiresAt *time.Time + // LadderRunID links this exchange record to the ladder run that created it. + // Nil for standalone ri_exchange_reshape task records. + // The database column ri_exchange_history.ladder_run_id was added in + // migration 000080 and is the authoritative source for origin scoping. + LadderRunID *string + CreatedAt time.Time + UpdatedAt time.Time + CompletedAt *time.Time + ExpiresAt *time.Time } // RIExchangeStore is the subset of store operations needed by RunAutoExchange. type RIExchangeStore interface { SaveRIExchangeRecord(ctx context.Context, record *ExchangeRecord) error + // CancelAllPendingExchanges cancels every pending record regardless of origin. + // Kept for interface compatibility; RunAutoExchange now calls + // CancelPendingExchangesByOrigin instead to avoid cross-origin contamination. CancelAllPendingExchanges(ctx context.Context) (int64, error) + // CancelPendingExchangesByOrigin cancels only pending records that match the + // given origin scope: + // - ladderScoped=false (standalone): cancels WHERE ladder_run_id IS NULL + // - ladderScoped=true (ladder): cancels WHERE ladder_run_id IS NOT NULL + // This prevents the standalone task from wiping out ladder-linked pending + // reshapes and vice versa. + CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]ExchangeRecord, error) GetRIExchangeDailySpend(ctx context.Context, date time.Time) (string, error) CompleteRIExchange(ctx context.Context, id string, exchangeID string) error @@ -77,6 +106,19 @@ type RunAutoExchangeParams struct { // RIMetadata maps RI ID to its metadata (product description, tenancy, scope, duration). RIMetadata map[string]RIMetadataInfo + + // LadderRunID links all exchange records created by this run to the ladder + // run that originated them. Nil for the standalone ri_exchange_reshape task. + // When set, RunAutoExchange calls CancelPendingExchangesByOrigin with + // ladderScoped=true (cancels only ladder-linked pendings). When nil, it + // cancels only standalone (ladder_run_id IS NULL) pendings. + LadderRunID *string + + // DryRun skips all state mutations (cancellations, record saves, exchange + // executions). Outcomes are returned with Simulated=true. No store write + // is performed; mock.AssertExpectations will catch any accidental mutation + // call in tests (fail-loud gate). + DryRun bool } // RIMetadataInfo holds the offering metadata for a specific RI. @@ -109,6 +151,9 @@ type ExchangeOutcome struct { ExchangeID string UtilizationPct float64 Error string + // Simulated is true when RunAutoExchangeParams.DryRun was set. The outcome + // was analyzed and quoted but no record was saved and no exchange was executed. + Simulated bool } // SkippedRecommendation captures a recommendation that was not processed. @@ -124,16 +169,22 @@ const staleProcessingThreshold = 15 * time.Minute func RunAutoExchange(ctx context.Context, params RunAutoExchangeParams) (*AutoExchangeResult, error) { result := &AutoExchangeResult{Mode: params.Config.Mode} - // 1. Cancel all stale pending records. + // 1. Cancel stale pending records with origin-scoped cancellation. + // DryRun skips ALL mutations including cancellation (no state is changed). + // Non-DryRun: cancel only pending records that share this run's origin so + // the standalone task does not wipe out ladder-linked pendings and vice versa. // Race condition note: if a user clicks approve at 5h59m while this new run // fires and cancels pending records, the TransitionRIExchangeStatus atomic // WHERE clause prevents the exchange from executing (record already cancelled // → returns nil → handler returns 409). - cancelled, err := params.Store.CancelAllPendingExchanges(ctx) - if err != nil { - logging.Warnf("failed to cancel pending exchanges: %v", err) - } else if cancelled > 0 { - logging.Infof("cancelled %d stale pending exchange records", cancelled) + if !params.DryRun { + ladderScoped := params.LadderRunID != nil + cancelled, err := params.Store.CancelPendingExchangesByOrigin(ctx, ladderScoped) + if err != nil { + logging.Warnf("failed to cancel pending exchanges: %v", err) + } else if cancelled > 0 { + logging.Infof("cancelled %d stale pending exchange records (ladder_scoped=%v)", cancelled, ladderScoped) + } } // 2. Log warning for stale processing records @@ -275,6 +326,22 @@ func getValidatedQuote(ctx context.Context, params RunAutoExchangeParams, rec Re } func processManualExchange(ctx context.Context, params RunAutoExchangeParams, rec ReshapeRecommendation, offeringID, paymentDueStr string) ExchangeOutcome { + // DryRun: return a simulated pending outcome without any record save or + // token generation (tokens are actionable money instruments — never issue + // live tokens in a dry run). + if params.DryRun { + return ExchangeOutcome{ + SourceRIID: rec.SourceRIID, + SourceInstanceType: rec.SourceInstanceType, + TargetInstanceType: rec.TargetInstanceType, + TargetOfferingID: offeringID, + TargetCount: rec.TargetCount, + PaymentDue: paymentDueStr, + UtilizationPct: rec.UtilizationPercent, + Simulated: true, + } + } + token, err := common.GenerateApprovalToken() if err != nil { logging.Errorf("failed to generate approval token for %s: %v", rec.SourceRIID, err) @@ -296,7 +363,7 @@ func processManualExchange(ctx context.Context, params RunAutoExchangeParams, re } } // 24h is a safety net; the email says "approve within 6 hours" because - // CancelAllPendingExchanges at the next run start (every 6h) will cancel + // CancelPendingExchangesByOrigin at the next run start (every 6h) will cancel // this record. The 24h expiry catches edge cases where the scheduled run // is delayed or disabled. expiresAt := time.Now().Add(24 * time.Hour) @@ -315,6 +382,7 @@ func processManualExchange(ctx context.Context, params RunAutoExchangeParams, re ApprovalToken: token, Mode: string(ExchangeModeManual), ExpiresAt: &expiresAt, + LadderRunID: params.LadderRunID, } if err := params.Store.SaveRIExchangeRecord(ctx, record); err != nil { @@ -360,6 +428,14 @@ func processAutoExchange(ctx context.Context, params RunAutoExchangeParams, rec UtilizationPct: rec.UtilizationPercent, } + // DryRun: return a simulated completed outcome without executing the exchange + // or saving any record. The daily cap is intentionally skipped so the + // simulation reflects what would happen if the cap were not a factor. + if params.DryRun { + outcome.Simulated = true + return outcome + } + // Check daily cap dailySpendStr, err := params.Store.GetRIExchangeDailySpend(ctx, time.Now()) if err != nil { @@ -412,7 +488,7 @@ func processAutoExchange(ctx context.Context, params RunAutoExchangeParams, rec return outcome } - // Save completed record + // Save completed record with ladder linkage when applicable. now := time.Now() record := &ExchangeRecord{ AccountID: params.AccountID, @@ -428,6 +504,7 @@ func processAutoExchange(ctx context.Context, params RunAutoExchangeParams, rec Status: "completed", Mode: string(ExchangeModeAuto), CompletedAt: &now, + LadderRunID: params.LadderRunID, } if err := params.Store.SaveRIExchangeRecord(ctx, record); err != nil { @@ -467,6 +544,7 @@ func saveFailedRecord(ctx context.Context, params RunAutoExchangeParams, rec Res Status: "failed", Error: errMsg, Mode: string(mode), + LadderRunID: params.LadderRunID, } if err := params.Store.SaveRIExchangeRecord(ctx, record); err != nil { logging.Errorf("failed to save failed exchange record for %s: %v", rec.SourceRIID, err) diff --git a/pkg/exchange/auto_test.go b/pkg/exchange/auto_test.go index 17451d1f4..c045fbf03 100644 --- a/pkg/exchange/auto_test.go +++ b/pkg/exchange/auto_test.go @@ -7,6 +7,7 @@ import ( "time" "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" ) @@ -17,6 +18,9 @@ type mockExchangeStore struct { staleRecords []ExchangeRecord dailySpend string dailySpendErr error + // cancelByOriginLastScoped captures the ladderScoped argument of the last + // CancelPendingExchangesByOrigin call for assertion in scoping tests. + cancelByOriginLastScoped *bool } func (m *mockExchangeStore) SaveRIExchangeRecord(_ context.Context, record *ExchangeRecord) error { @@ -31,6 +35,11 @@ func (m *mockExchangeStore) CancelAllPendingExchanges(_ context.Context) (int64, return m.cancelledCount, nil } +func (m *mockExchangeStore) CancelPendingExchangesByOrigin(_ context.Context, ladderScoped bool) (int64, error) { + m.cancelByOriginLastScoped = &ladderScoped + return m.cancelledCount, nil +} + func (m *mockExchangeStore) GetStaleProcessingExchanges(_ context.Context, _ time.Duration) ([]ExchangeRecord, error) { return m.staleRecords, nil } @@ -50,6 +59,51 @@ func (m *mockExchangeStore) FailRIExchange(_ context.Context, _ string, _ string return nil } +// testifyExchangeStore is a testify mock.Mock-based store for tests that need +// mock.AssertExpectations enforcement (e.g. the dry-run test that must assert +// zero mutations were attempted). +type testifyExchangeStore struct { + mock.Mock +} + +func (m *testifyExchangeStore) SaveRIExchangeRecord(ctx context.Context, record *ExchangeRecord) error { + args := m.Called(ctx, record) + return args.Error(0) +} + +func (m *testifyExchangeStore) CancelAllPendingExchanges(ctx context.Context) (int64, error) { + args := m.Called(ctx) + return args.Get(0).(int64), args.Error(1) +} + +func (m *testifyExchangeStore) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { + args := m.Called(ctx, ladderScoped) + return args.Get(0).(int64), args.Error(1) +} + +func (m *testifyExchangeStore) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]ExchangeRecord, error) { + args := m.Called(ctx, olderThan) + if args.Get(0) == nil { + return nil, args.Error(1) + } + return args.Get(0).([]ExchangeRecord), args.Error(1) +} + +func (m *testifyExchangeStore) GetRIExchangeDailySpend(ctx context.Context, date time.Time) (string, error) { + args := m.Called(ctx, date) + return args.String(0), args.Error(1) +} + +func (m *testifyExchangeStore) CompleteRIExchange(ctx context.Context, id string, exchangeID string) error { + args := m.Called(ctx, id, exchangeID) + return args.Error(0) +} + +func (m *testifyExchangeStore) FailRIExchange(ctx context.Context, id string, errMsg string) error { + args := m.Called(ctx, id, errMsg) + return args.Error(0) +} + // mockExchangeClient implements ExchangeClientInterface for testing. type mockExchangeClient struct { quoteResult *ExchangeQuoteSummary @@ -310,3 +364,151 @@ func TestRunAutoExchange_IdleRI_Skipped(t *testing.T) { assert.Len(t, result.Skipped, 1) assert.Contains(t, result.Skipped[0].Reason, "idle") } + +// ─── Scoped cancellation tests ──────────────────────────────────────────────── + +// TestRunAutoExchange_StandaloneDoesNotCancelLadderPendings is the regression +// test for gap G10: before this fix, RunAutoExchange called +// CancelAllPendingExchanges (store-wide), so the standalone ri_exchange_reshape +// task would silently wipe out ladder-linked pending reshapes. After the fix, +// the standalone run calls CancelPendingExchangesByOrigin(ctx, false) which +// only cancels records where ladder_run_id IS NULL. +// +// Regression: pre-fix code routes through CancelAllPendingExchanges and would +// cancel with ladderScoped=nil (method absent). Post-fix: CancelPendingExchangesByOrigin +// is called with ladderScoped=false. +func TestRunAutoExchange_StandaloneDoesNotCancelLadderPendings(t *testing.T) { + t.Parallel() + store := &mockExchangeStore{dailySpend: "0"} + client := &mockExchangeClient{quoteResult: defaultQuote()} + params := defaultParams(store, client) + // LadderRunID=nil means standalone origin. + params.LadderRunID = nil + + _, err := RunAutoExchange(context.Background(), params) + require.NoError(t, err) + + // Must have called the scoped cancel with ladderScoped=false, NOT the + // store-wide CancelAllPendingExchanges. cancelByOriginLastScoped tracks the + // argument passed to CancelPendingExchangesByOrigin. + require.NotNil(t, store.cancelByOriginLastScoped, + "CancelPendingExchangesByOrigin must have been called (not the store-wide variant)") + assert.False(t, *store.cancelByOriginLastScoped, + "standalone run must cancel only non-ladder pendings (ladderScoped=false)") +} + +// TestRunAutoExchange_LadderDoesNotCancelStandalonePendings verifies that a +// ladder-originated run calls CancelPendingExchangesByOrigin(ctx, true), which +// only cancels records where ladder_run_id IS NOT NULL, leaving standalone +// pending records intact. +func TestRunAutoExchange_LadderDoesNotCancelStandalonePendings(t *testing.T) { + t.Parallel() + store := &mockExchangeStore{dailySpend: "0"} + client := &mockExchangeClient{quoteResult: defaultQuote()} + params := defaultParams(store, client) + ladderRunID := "ladder-run-abc-123" + params.LadderRunID = &ladderRunID + + _, err := RunAutoExchange(context.Background(), params) + require.NoError(t, err) + + require.NotNil(t, store.cancelByOriginLastScoped, + "CancelPendingExchangesByOrigin must have been called") + assert.True(t, *store.cancelByOriginLastScoped, + "ladder run must cancel only ladder-linked pendings (ladderScoped=true)") +} + +// ─── Dry-run tests ──────────────────────────────────────────────────────────── + +// TestRunAutoExchange_DryRun_ManualMode_ZeroMutations verifies that a dry-run +// in manual mode produces simulated pending outcomes without any store writes +// or cancellation calls. mock.AssertExpectations enforces the zero-mutation +// contract (fail-loud gate: any registered-but-uncalled or called-but-unregistered +// method on a testify mock fails the test). +func TestRunAutoExchange_DryRun_ManualMode_ZeroMutations(t *testing.T) { + t.Parallel() + store := &testifyExchangeStore{} + // Only GetStaleProcessingExchanges is a read-only call expected in dry-run. + store.On("GetStaleProcessingExchanges", mock.Anything, staleProcessingThreshold). + Return(nil, nil) + t.Cleanup(func() { store.AssertExpectations(t) }) + + client := &mockExchangeClient{quoteResult: defaultQuote()} + params := defaultParams(store, client) + params.DryRun = true + + result, err := RunAutoExchange(context.Background(), params) + require.NoError(t, err) + + // Outcome must be simulated — no real token, no record ID. + require.Len(t, result.Pending, 1) + assert.True(t, result.Pending[0].Simulated, "outcome must be tagged Simulated in dry-run") + assert.Empty(t, result.Pending[0].ApprovalToken, "no live approval token must be generated in dry-run") + assert.Empty(t, result.Pending[0].RecordID, "no record must be saved in dry-run") +} + +// TestRunAutoExchange_DryRun_AutoMode_ZeroMutations verifies that a dry-run +// in auto mode produces simulated completed outcomes without executing any +// exchange or saving any record. +func TestRunAutoExchange_DryRun_AutoMode_ZeroMutations(t *testing.T) { + t.Parallel() + store := &testifyExchangeStore{} + store.On("GetStaleProcessingExchanges", mock.Anything, staleProcessingThreshold). + Return(nil, nil) + t.Cleanup(func() { store.AssertExpectations(t) }) + + client := &mockExchangeClient{quoteResult: defaultQuote(), executeResult: "exch-dryrun-ignored"} + params := defaultParams(store, client) + params.Config.Mode = "auto" + params.DryRun = true + + result, err := RunAutoExchange(context.Background(), params) + require.NoError(t, err) + + // Auto mode dry-run lands in Completed (would have executed) but Simulated. + require.Len(t, result.Completed, 1) + assert.True(t, result.Completed[0].Simulated, "auto dry-run outcome must be tagged Simulated") + assert.Empty(t, result.Completed[0].ExchangeID, "no real exchange must be executed in dry-run") +} + +// ─── LadderRunID round-trip test ────────────────────────────────────────────── + +// TestRunAutoExchange_LadderRunID_StampedOnRecords verifies that when +// RunAutoExchangeParams.LadderRunID is set, every saved record (pending, +// completed, failed) carries that ID so the DB column is actually written. +func TestRunAutoExchange_LadderRunID_StampedOnRecords(t *testing.T) { + t.Parallel() + store := &mockExchangeStore{dailySpend: "0"} + client := &mockExchangeClient{quoteResult: defaultQuote()} + params := defaultParams(store, client) + ladderRunID := "ladder-run-999" + params.LadderRunID = &ladderRunID + + result, err := RunAutoExchange(context.Background(), params) + require.NoError(t, err) + + require.Len(t, result.Pending, 1, "expected one pending record in manual mode") + require.Len(t, store.savedRecords, 1) + require.NotNil(t, store.savedRecords[0].LadderRunID, "LadderRunID must be non-nil") + assert.Equal(t, ladderRunID, *store.savedRecords[0].LadderRunID, + "saved record must carry the LadderRunID from params") +} + +// TestRunAutoExchange_NoLadderRunID_RecordHasNilLadderRunID verifies that +// standalone (LadderRunID=nil) runs save records with nil LadderRunID, so +// the DB column remains NULL and CancelPendingExchangesByOrigin(false) can +// correctly target them. +func TestRunAutoExchange_NoLadderRunID_RecordHasNilLadderRunID(t *testing.T) { + t.Parallel() + store := &mockExchangeStore{dailySpend: "0"} + client := &mockExchangeClient{quoteResult: defaultQuote()} + params := defaultParams(store, client) + params.LadderRunID = nil + + _, err := RunAutoExchange(context.Background(), params) + require.NoError(t, err) + + require.Len(t, store.savedRecords, 1) + assert.Nil(t, store.savedRecords[0].LadderRunID, + "standalone run must save record with nil LadderRunID") +} diff --git a/pkg/go.mod b/pkg/go.mod index 28393ba32..90de1d9a5 100644 --- a/pkg/go.mod +++ b/pkg/go.mod @@ -29,4 +29,5 @@ require ( github.com/aws/smithy-go v1.24.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/stretchr/objx v0.5.2 // indirect ) diff --git a/pkg/go.sum b/pkg/go.sum index a03f6a921..40ea94bc2 100644 --- a/pkg/go.sum +++ b/pkg/go.sum @@ -34,6 +34,8 @@ github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZb github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA= github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= diff --git a/providers/aws/ladder/reshape.go b/providers/aws/ladder/reshape.go index 05d6947b7..0e2e69a70 100644 --- a/providers/aws/ladder/reshape.go +++ b/providers/aws/ladder/reshape.go @@ -28,20 +28,19 @@ const unlimitedCapUSD = math.MaxFloat64 // store, quote/execute client, offering lookup, and RI/utilization inventory // (the same wiring internal/server.executeRIExchangeReshape performs). // -// WARNING — store-wide side effect: the underlying exchange.RunAutoExchange -// begins by unconditionally canceling ALL pending exchange records in the -// store (CancelAllPendingExchanges), including pendings created by the -// standalone ri_exchange_reshape scheduled task. Callers MUST NOT run -// ReshapeBuffer concurrently with that task against the same store; the -// pipeline phase that invokes this method must coordinate with (or disable) -// the standalone scheduler. +// Origin-scoped cancellation: the underlying exchange.RunAutoExchange now +// calls CancelPendingExchangesByOrigin(ctx, true) — cancelling only +// ladder-linked pending records (ladder_run_id IS NOT NULL) — rather than the +// former store-wide CancelAllPendingExchanges. This means the standalone +// ri_exchange_reshape task's pending records are no longer affected by a ladder +// reshape run (gap G10, issue #1348). // -// DryRun is NOT supported: there is no true simulation mode in pkg/exchange -// yet (tracked upstream), and mapping DryRun onto the exchange flow's manual -// mode would be a false simulation — manual mode persists pending -// ExchangeRecords with live approval tokens (actionable money instruments) -// and still triggers the store-wide pending cancellation above. cfg.DryRun -// therefore fails loud; the engine previews reshapes via ActionReshape +// DryRun is NOT yet wired end-to-end in this method: pkg/exchange now supports +// true dry-run (RunAutoExchangeParams.DryRun=true skips all mutations and +// returns Simulated outcomes), but passing cfg.DryRun through to the exchange +// runner requires the exchange runner parameter to be surfaced in the +// ReshapeRunner interface, which is tracked in L16. Until L16 lands, +// cfg.DryRun fails loud here; the engine previews reshapes via ActionReshape // rationales without calling ReshapeBuffer. // // Config mapping (BufferReshapeConfig -> exchange.RIExchangeConfig): From 5b9a8e525e7462705284cd405cfe725a1ee895b0 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 16 Jul 2026 12:21:42 +0300 Subject: [PATCH 2/2] fix(exchange): typed origin enum + correct-by-construction ladder seam (L3 review) Address Fable FIX-THEN-SHIP findings on the L3 diff: 1. [HIGH] Make ladder-side scoping reachable and honest. Widen the exchangeRunner.RunAutoExchange seam to carry (ladderRunID *string, dryRun bool) as explicit params, add BufferReshapeConfig.LadderRunID, and have ReshapeBuffer forward both. The concrete runner (L16) is now FORCED to thread LadderRunID, so it cannot default to nil and cancel standalone pendings (the G10 bug). Correct the reshape.go comments: standalone scoping is active now; ladder-side scoping activates when the runner is wired (L16). Remove the stale "dry-run not supported" rejection; DryRun is forwarded. 3. [MEDIUM] Replace the bare `ladderScoped bool` with a typed common.ExchangeOrigin enum end-to-end (exchange.RIExchangeStore, config.StoreInterface, PostgresStore, adapter, all mocks). The enum moves to pkg/common (no heavy SDK deps) so internal/config can import it. PostgresStore validates the origin and fails loud on an unknown value. 2. [MEDIUM] Document the deliberate coarse IS NOT NULL partition (cancels all ladder-linked pendings, not per-run) with a TODO(#1367) at the SQL. 4. [MEDIUM] Tighten the pgxmock branch tests to match `ladder_run_id IS NULL` vs `IS NOT NULL` (mutually exclusive regexes) so a branch mix-up fails; add an unknown-origin fail-loud test and a 21-arg SaveRIExchangeRecord INSERT column-alignment test. 5. [LOW] Document the FK ON DELETE SET NULL reclassification (deleting a ladder run makes its still-pending reshape standalone-cancellable) as known/acceptable. 6. [NIT] Touch updated_at in the scoped cancel queries for parity. Verified fail-before/pass-after: swapping the two WHERE clauses fails both TestPGXMock_CancelPendingExchangesByOrigin_{Standalone,Ladder}; correct SQL passes. --- internal/analytics/collector_test.go | 3 +- internal/config/interfaces.go | 12 ++-- internal/config/store_postgres.go | 35 +++++++--- internal/config/store_postgres_nil_db_test.go | 6 +- .../config/store_postgres_pgxmock_test.go | 58 ++++++++++++++-- internal/config/types.go | 7 ++ internal/mocks/stores.go | 5 +- internal/server/handler_ri_exchange.go | 5 +- internal/server/handler_ri_exchange_test.go | 7 +- internal/server/test_helpers_test.go | 3 +- pkg/common/types.go | 34 +++++++++ pkg/exchange/auto.go | 36 ++++------ pkg/exchange/auto_test.go | 45 ++++++------ pkg/ladder/capability.go | 10 +++ providers/aws/ladder/interfaces.go | 12 +++- providers/aws/ladder/reshape.go | 49 +++++++------ providers/aws/ladder/reshape_test.go | 69 +++++++++++++++---- 17 files changed, 279 insertions(+), 117 deletions(-) diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index ed989b6c2..ff7785a72 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -8,6 +8,7 @@ import ( "time" "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/pkg/common" "github.com/LeanerCloud/CUDly/pkg/ladder" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/assert" @@ -321,7 +322,7 @@ func (m *mockConfigStore) CancelAllPendingExchanges(ctx context.Context) (int64, return 0, nil } -func (m *mockConfigStore) CancelPendingExchangesByOrigin(_ context.Context, _ bool) (int64, error) { +func (m *mockConfigStore) CancelPendingExchangesByOrigin(_ context.Context, _ common.ExchangeOrigin) (int64, error) { return 0, nil } func (m *mockConfigStore) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]config.RIExchangeRecord, error) { diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index b75a98a5b..c22f6bdc5 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -6,6 +6,7 @@ import ( "github.com/jackc/pgx/v5" + "github.com/LeanerCloud/CUDly/pkg/common" "github.com/LeanerCloud/CUDly/pkg/ladder" ) @@ -209,13 +210,14 @@ type StoreInterface interface { FailRIExchange(ctx context.Context, id string, errorMsg string) error GetRIExchangeDailySpend(ctx context.Context, date time.Time) (string, error) CancelAllPendingExchanges(ctx context.Context) (int64, error) - // CancelPendingExchangesByOrigin cancels only pending records that match the - // given origin scope: - // - ladderScoped=false (standalone): cancels WHERE ladder_run_id IS NULL - // - ladderScoped=true (ladder): cancels WHERE ladder_run_id IS NOT NULL + // CancelPendingExchangesByOrigin cancels only pending records whose origin + // matches: + // - common.ExchangeOriginStandalone: cancels WHERE ladder_run_id IS NULL + // - common.ExchangeOriginLadder: cancels WHERE ladder_run_id IS NOT NULL + // The origin is validated at the boundary and an unknown value fails loud. // This prevents the standalone ri_exchange_reshape task from wiping out // ladder-linked pending reshapes and vice versa (gap G10 / issue #1348). - CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) + CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]RIExchangeRecord, error) // Cloud accounts diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 3b7754fc0..4316f9537 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -12,6 +12,7 @@ import ( "time" "github.com/LeanerCloud/CUDly/internal/database" + "github.com/LeanerCloud/CUDly/pkg/common" "github.com/google/uuid" "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgconn" @@ -2452,22 +2453,38 @@ func (s *PostgresStore) CancelAllPendingExchanges(ctx context.Context) (int64, e } // CancelPendingExchangesByOrigin cancels only pending records that match the -// given origin scope (gap G10 / issue #1348): -// - ladderScoped=false (standalone): cancels WHERE ladder_run_id IS NULL -// - ladderScoped=true (ladder): cancels WHERE ladder_run_id IS NOT NULL -func (s *PostgresStore) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { +// given origin (gap G10 / issue #1348): +// - common.ExchangeOriginStandalone: cancels WHERE ladder_run_id IS NULL +// - common.ExchangeOriginLadder: cancels WHERE ladder_run_id IS NOT NULL +// +// The origin is validated at this boundary; an unknown value fails loud rather +// than silently cancelling the wrong partition on a money path. +// +// DELIBERATE COARSE PARTITION: the ExchangeOriginLadder branch cancels EVERY +// ladder-linked pending record (ladder_run_id IS NOT NULL) across ALL ladder +// runs and configs, not just the current run's. This is acceptable today +// because the ladder never creates pending exchange records: buildRIExchangeConfig +// forces Mode=auto, which completes or fails immediately without leaving a +// pending row. Per-run / per-config scoping needs an additional selection key +// (e.g. the specific ladder_run_id or config_id) and is tracked in TODO(#1367). +func (s *PostgresStore) CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) { + if err := origin.Validate(); err != nil { + return 0, fmt.Errorf("CancelPendingExchangesByOrigin: %w", err) + } + var query string - if ladderScoped { + switch origin { + case common.ExchangeOriginLadder: query = ` UPDATE ri_exchange_history - SET status = 'cancelled' + SET status = 'cancelled', updated_at = NOW() WHERE status = 'pending' AND ladder_run_id IS NOT NULL ` - } else { + default: // common.ExchangeOriginStandalone (validated non-unknown above) query = ` UPDATE ri_exchange_history - SET status = 'cancelled' + SET status = 'cancelled', updated_at = NOW() WHERE status = 'pending' AND ladder_run_id IS NULL ` @@ -2475,7 +2492,7 @@ func (s *PostgresStore) CancelPendingExchangesByOrigin(ctx context.Context, ladd result, err := s.db.Exec(ctx, query) if err != nil { - return 0, fmt.Errorf("failed to cancel pending exchanges by origin (ladder_scoped=%v): %w", ladderScoped, err) + return 0, fmt.Errorf("failed to cancel pending exchanges by origin %q: %w", origin, err) } return result.RowsAffected(), nil diff --git a/internal/config/store_postgres_nil_db_test.go b/internal/config/store_postgres_nil_db_test.go index e52247c4e..5cd6b96df 100644 --- a/internal/config/store_postgres_nil_db_test.go +++ b/internal/config/store_postgres_nil_db_test.go @@ -6,6 +6,8 @@ import ( "time" "github.com/stretchr/testify/assert" + + "github.com/LeanerCloud/CUDly/pkg/common" ) // TestPostgresStore_SaveRIExchangeRecord_GeneratesID verifies that SaveRIExchangeRecord @@ -170,8 +172,10 @@ func TestPostgresStore_CancelPendingExchangesByOrigin_NilDB(t *testing.T) { store := NewPostgresStore(nil) ctx := context.Background() + // Pass a VALID origin so the method reaches the nil db (an invalid origin + // would fail loud at Validate before touching the connection). panicked := callWithRecover(func() { - _, _ = store.CancelPendingExchangesByOrigin(ctx, false) + _, _ = store.CancelPendingExchangesByOrigin(ctx, common.ExchangeOriginStandalone) }) assert.True(t, panicked, "expected panic with nil db connection") diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index 795319015..cb5b1ef44 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -15,6 +15,8 @@ import ( "github.com/pashagolub/pgxmock/v4" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/CUDly/pkg/common" ) // newMock creates a pgxmock pool with regexp query matching. @@ -1819,17 +1821,22 @@ func TestPGXMock_CancelAllPendingExchanges_Success(t *testing.T) { // ─── CancelPendingExchangesByOrigin ────────────────────────────────────────── +// The regex pins each test to its own WHERE clause so a branch mix-up (both +// branches emitting the same SQL) fails the test. "ladder_run_id IS NULL" is +// NOT a substring of "ladder_run_id IS NOT NULL", so the two matchers are +// mutually exclusive. func TestPGXMock_CancelPendingExchangesByOrigin_Standalone(t *testing.T) { mock := newMock(t) store := storeWith(mock) ctx := context.Background() - // ladderScoped=false -> WHERE ladder_run_id IS NULL - mock.ExpectExec("UPDATE").WillReturnResult(pgxmock.NewResult("UPDATE", 2)) + // Standalone origin must target WHERE ladder_run_id IS NULL only. + mock.ExpectExec("ladder_run_id IS NULL").WillReturnResult(pgxmock.NewResult("UPDATE", 2)) - n, err := store.CancelPendingExchangesByOrigin(ctx, false) + n, err := store.CancelPendingExchangesByOrigin(ctx, common.ExchangeOriginStandalone) require.NoError(t, err) assert.Equal(t, int64(2), n) + assert.NoError(t, mock.ExpectationsWereMet()) } func TestPGXMock_CancelPendingExchangesByOrigin_Ladder(t *testing.T) { @@ -1837,12 +1844,51 @@ func TestPGXMock_CancelPendingExchangesByOrigin_Ladder(t *testing.T) { store := storeWith(mock) ctx := context.Background() - // ladderScoped=true -> WHERE ladder_run_id IS NOT NULL - mock.ExpectExec("UPDATE").WillReturnResult(pgxmock.NewResult("UPDATE", 1)) + // Ladder origin must target WHERE ladder_run_id IS NOT NULL only. + mock.ExpectExec("ladder_run_id IS NOT NULL").WillReturnResult(pgxmock.NewResult("UPDATE", 1)) - n, err := store.CancelPendingExchangesByOrigin(ctx, true) + n, err := store.CancelPendingExchangesByOrigin(ctx, common.ExchangeOriginLadder) require.NoError(t, err) assert.Equal(t, int64(1), n) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestPGXMock_CancelPendingExchangesByOrigin_UnknownOrigin_FailsLoud(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + ctx := context.Background() + + // An unknown origin must fail at the boundary WITHOUT issuing any query + // (no ExpectExec registered → ExpectationsWereMet stays satisfied). + n, err := store.CancelPendingExchangesByOrigin(ctx, common.ExchangeOrigin("bogus")) + require.Error(t, err) + assert.Contains(t, err.Error(), "unknown exchange origin") + assert.Equal(t, int64(0), n) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +// ─── SaveRIExchangeRecord ───────────────────────────────────────────────────── + +func TestPGXMock_SaveRIExchangeRecord_InsertColumnAlignment(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + ctx := context.Background() + + // 21 columns/placeholders: original 20 + ladder_run_id ($21, issue #1348). + // A column/placeholder count drift makes WithArgs(anyArgsCfg(21)) fail. + mock.ExpectExec("INSERT INTO ri_exchange_history").WithArgs(anyArgsCfg(21)...). + WillReturnResult(pgxmock.NewResult("INSERT", 1)) + + ladderRunID := "run-123" + err := store.SaveRIExchangeRecord(ctx, &RIExchangeRecord{ + AccountID: "acc-1", + Region: "us-east-1", + Status: "pending", + Mode: "manual", + LadderRunID: &ladderRunID, + }) + require.NoError(t, err) + assert.NoError(t, mock.ExpectationsWereMet()) } // ─── CleanupOldExecutions ───────────────────────────────────────────────────── diff --git a/internal/config/types.go b/internal/config/types.go index f0b93cb74..e895b872c 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -810,6 +810,13 @@ type RIExchangeRecord struct { // The database column ri_exchange_history.ladder_run_id was added in migration // 000080 and is the authoritative source for origin scoping in // CancelPendingExchangesByOrigin. + // + // KNOWN/ACCEPTABLE: the FK is ON DELETE SET NULL (migration 000080), so + // deleting a ladder_runs row nulls this column and reclassifies the record + // as standalone. A still-pending reshape then becomes standalone-cancellable + // (the standalone-origin sweep would cancel it). This is acceptable: a + // deleted run has no owner to approve its pendings, so cancelling them on the + // next standalone sweep is the safe outcome, not a leak. LadderRunID *string `json:"ladder_run_id,omitempty"` CreatedAt time.Time `json:"created_at"` UpdatedAt time.Time `json:"updated_at"` diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index fa263fb58..89e3077c2 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -7,6 +7,7 @@ import ( "github.com/LeanerCloud/CUDly/internal/auth" "github.com/LeanerCloud/CUDly/internal/config" + "github.com/LeanerCloud/CUDly/pkg/common" "github.com/LeanerCloud/CUDly/pkg/ladder" "github.com/jackc/pgx/v5" "github.com/stretchr/testify/mock" @@ -526,8 +527,8 @@ func (m *MockConfigStore) CancelAllPendingExchanges(ctx context.Context) (int64, return v, args.Error(1) } -func (m *MockConfigStore) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { - args := m.Called(ctx, ladderScoped) +func (m *MockConfigStore) CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) { + args := m.Called(ctx, origin) v, ok := args.Get(0).(int64) if !ok { panic(fmt.Sprintf("mock: expected int64, got %T", args.Get(0))) diff --git a/internal/server/handler_ri_exchange.go b/internal/server/handler_ri_exchange.go index fb9713781..087e977c1 100644 --- a/internal/server/handler_ri_exchange.go +++ b/internal/server/handler_ri_exchange.go @@ -12,6 +12,7 @@ import ( "github.com/LeanerCloud/CUDly/internal/config" "github.com/LeanerCloud/CUDly/internal/email" + "github.com/LeanerCloud/CUDly/pkg/common" "github.com/LeanerCloud/CUDly/pkg/exchange" awsprovider "github.com/LeanerCloud/CUDly/providers/aws" "github.com/LeanerCloud/CUDly/providers/aws/recommendations" @@ -294,8 +295,8 @@ func (a *configExchangeStoreAdapter) CancelAllPendingExchanges(ctx context.Conte return a.store.CancelAllPendingExchanges(ctx) } -func (a *configExchangeStoreAdapter) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { - return a.store.CancelPendingExchangesByOrigin(ctx, ladderScoped) +func (a *configExchangeStoreAdapter) CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) { + return a.store.CancelPendingExchangesByOrigin(ctx, origin) } func (a *configExchangeStoreAdapter) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]exchange.ExchangeRecord, error) { diff --git a/internal/server/handler_ri_exchange_test.go b/internal/server/handler_ri_exchange_test.go index fe7106577..2525f94ef 100644 --- a/internal/server/handler_ri_exchange_test.go +++ b/internal/server/handler_ri_exchange_test.go @@ -11,6 +11,7 @@ import ( "github.com/LeanerCloud/CUDly/internal/config" "github.com/LeanerCloud/CUDly/internal/email" "github.com/LeanerCloud/CUDly/internal/testutil" + "github.com/LeanerCloud/CUDly/pkg/common" "github.com/LeanerCloud/CUDly/pkg/exchange" "github.com/LeanerCloud/CUDly/providers/aws/recommendations" ec2svc "github.com/LeanerCloud/CUDly/providers/aws/services/ec2" @@ -795,7 +796,7 @@ type mockConfigStoreForExchange struct { globalConfigErr error saveRIExchangeRecordFunc func(ctx context.Context, record *config.RIExchangeRecord) error cancelAllPendingFunc func(ctx context.Context) (int64, error) - cancelPendingByOriginFunc func(ctx context.Context, ladderScoped bool) (int64, error) + cancelPendingByOriginFunc func(ctx context.Context, origin common.ExchangeOrigin) (int64, error) getDailySpendFunc func(ctx context.Context, date time.Time) (string, error) } @@ -823,9 +824,9 @@ func (m *mockConfigStoreForExchange) CancelAllPendingExchanges(ctx context.Conte return 0, nil } -func (m *mockConfigStoreForExchange) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { +func (m *mockConfigStoreForExchange) CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) { if m.cancelPendingByOriginFunc != nil { - return m.cancelPendingByOriginFunc(ctx, ladderScoped) + return m.cancelPendingByOriginFunc(ctx, origin) } return 0, nil } diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index e0e2901e0..fbc7843a3 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -6,6 +6,7 @@ import ( "github.com/LeanerCloud/CUDly/internal/config" "github.com/LeanerCloud/CUDly/internal/database" + "github.com/LeanerCloud/CUDly/pkg/common" "github.com/LeanerCloud/CUDly/pkg/ladder" "github.com/jackc/pgx/v5" ) @@ -171,7 +172,7 @@ func (m *mockConfigStoreForHealth) CancelAllPendingExchanges(ctx context.Context return 0, nil } -func (m *mockConfigStoreForHealth) CancelPendingExchangesByOrigin(_ context.Context, _ bool) (int64, error) { +func (m *mockConfigStoreForHealth) CancelPendingExchangesByOrigin(_ context.Context, _ common.ExchangeOrigin) (int64, error) { return 0, nil } func (m *mockConfigStoreForHealth) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]config.RIExchangeRecord, error) { diff --git a/pkg/common/types.go b/pkg/common/types.go index f6df840c3..9a494611d 100644 --- a/pkg/common/types.go +++ b/pkg/common/types.go @@ -26,6 +26,40 @@ const ( ProviderGCP ProviderType = "gcp" ) +// ExchangeOrigin identifies the task that created an RI exchange record so that +// pending-cancellation can be scoped by origin (standalone vs ladder) through a +// typed value rather than a bare bool. It lives in pkg/common because both +// pkg/exchange (the caller) and internal/config (the store implementation) need +// it, and pkg/common carries no heavy provider-SDK dependencies. +type ExchangeOrigin string + +const ( + // ExchangeOriginStandalone is the standalone ri_exchange_reshape scheduled + // task. Its pending records have ladder_run_id IS NULL in the database. + ExchangeOriginStandalone ExchangeOrigin = "standalone-ri-exchange" + // ExchangeOriginLadder is the cudly-ladder engine (ladder run reshape + // phase). Its pending records have ladder_run_id IS NOT NULL. + ExchangeOriginLadder ExchangeOrigin = "cudly-ladder" +) + +// Validate returns an error when the origin is not a known value. Callers on +// the money path (pending cancellation) must fail loud on an unexpected origin +// rather than silently defaulting to the wrong partition. +func (o ExchangeOrigin) Validate() error { + switch o { + case ExchangeOriginStandalone, ExchangeOriginLadder: + return nil + default: + return fmt.Errorf("unknown exchange origin %q (allowed: %s, %s)", + o, ExchangeOriginStandalone, ExchangeOriginLadder) + } +} + +// String returns the string representation of the exchange origin. +func (o ExchangeOrigin) String() string { + return string(o) +} + // String returns the string representation of the provider type func (p ProviderType) String() string { return string(p) diff --git a/pkg/exchange/auto.go b/pkg/exchange/auto.go index 41c9e38b5..b53be2f1f 100644 --- a/pkg/exchange/auto.go +++ b/pkg/exchange/auto.go @@ -10,20 +10,6 @@ import ( "github.com/LeanerCloud/CUDly/pkg/logging" ) -// ExchangeOrigin identifies the originating task that created an exchange record, -// enabling scoped cancellation so standalone and ladder-originated pending -// exchanges do not contaminate each other. -type ExchangeOrigin string - -const ( - // ExchangeOriginStandalone is the standalone ri_exchange_reshape scheduled task. - // Its pending records have ladder_run_id IS NULL in the database. - ExchangeOriginStandalone ExchangeOrigin = "standalone-ri-exchange" - // ExchangeOriginLadder is the cudly-ladder engine (ladder run reshape phase). - // Its pending records have ladder_run_id IS NOT NULL in the database. - ExchangeOriginLadder ExchangeOrigin = "cudly-ladder" -) - // ExchangeRecord is a lightweight record type for the auto exchange logic. // It mirrors config.RIExchangeRecord but lives in pkg/exchange to avoid // cross-module imports (pkg/ is a separate Go module from internal/). @@ -61,13 +47,14 @@ type RIExchangeStore interface { // Kept for interface compatibility; RunAutoExchange now calls // CancelPendingExchangesByOrigin instead to avoid cross-origin contamination. CancelAllPendingExchanges(ctx context.Context) (int64, error) - // CancelPendingExchangesByOrigin cancels only pending records that match the - // given origin scope: - // - ladderScoped=false (standalone): cancels WHERE ladder_run_id IS NULL - // - ladderScoped=true (ladder): cancels WHERE ladder_run_id IS NOT NULL + // CancelPendingExchangesByOrigin cancels only pending records whose origin + // matches: + // - common.ExchangeOriginStandalone: cancels WHERE ladder_run_id IS NULL + // - common.ExchangeOriginLadder: cancels WHERE ladder_run_id IS NOT NULL // This prevents the standalone task from wiping out ladder-linked pending - // reshapes and vice versa. - CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) + // reshapes and vice versa. Implementations must validate the origin and + // fail loud on an unknown value (money path). + CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) GetStaleProcessingExchanges(ctx context.Context, olderThan time.Duration) ([]ExchangeRecord, error) GetRIExchangeDailySpend(ctx context.Context, date time.Time) (string, error) CompleteRIExchange(ctx context.Context, id string, exchangeID string) error @@ -178,12 +165,15 @@ func RunAutoExchange(ctx context.Context, params RunAutoExchangeParams) (*AutoEx // WHERE clause prevents the exchange from executing (record already cancelled // → returns nil → handler returns 409). if !params.DryRun { - ladderScoped := params.LadderRunID != nil - cancelled, err := params.Store.CancelPendingExchangesByOrigin(ctx, ladderScoped) + origin := common.ExchangeOriginStandalone + if params.LadderRunID != nil { + origin = common.ExchangeOriginLadder + } + cancelled, err := params.Store.CancelPendingExchangesByOrigin(ctx, origin) if err != nil { logging.Warnf("failed to cancel pending exchanges: %v", err) } else if cancelled > 0 { - logging.Infof("cancelled %d stale pending exchange records (ladder_scoped=%v)", cancelled, ladderScoped) + logging.Infof("cancelled %d stale pending exchange records (origin=%s)", cancelled, origin) } } diff --git a/pkg/exchange/auto_test.go b/pkg/exchange/auto_test.go index c045fbf03..babaaea6e 100644 --- a/pkg/exchange/auto_test.go +++ b/pkg/exchange/auto_test.go @@ -9,6 +9,8 @@ import ( "github.com/stretchr/testify/assert" "github.com/stretchr/testify/mock" "github.com/stretchr/testify/require" + + "github.com/LeanerCloud/CUDly/pkg/common" ) // mockExchangeStore implements RIExchangeStore for testing. @@ -18,9 +20,9 @@ type mockExchangeStore struct { staleRecords []ExchangeRecord dailySpend string dailySpendErr error - // cancelByOriginLastScoped captures the ladderScoped argument of the last + // cancelByOriginLast captures the origin argument of the last // CancelPendingExchangesByOrigin call for assertion in scoping tests. - cancelByOriginLastScoped *bool + cancelByOriginLast *common.ExchangeOrigin } func (m *mockExchangeStore) SaveRIExchangeRecord(_ context.Context, record *ExchangeRecord) error { @@ -35,8 +37,8 @@ func (m *mockExchangeStore) CancelAllPendingExchanges(_ context.Context) (int64, return m.cancelledCount, nil } -func (m *mockExchangeStore) CancelPendingExchangesByOrigin(_ context.Context, ladderScoped bool) (int64, error) { - m.cancelByOriginLastScoped = &ladderScoped +func (m *mockExchangeStore) CancelPendingExchangesByOrigin(_ context.Context, origin common.ExchangeOrigin) (int64, error) { + m.cancelByOriginLast = &origin return m.cancelledCount, nil } @@ -76,8 +78,8 @@ func (m *testifyExchangeStore) CancelAllPendingExchanges(ctx context.Context) (i return args.Get(0).(int64), args.Error(1) } -func (m *testifyExchangeStore) CancelPendingExchangesByOrigin(ctx context.Context, ladderScoped bool) (int64, error) { - args := m.Called(ctx, ladderScoped) +func (m *testifyExchangeStore) CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) { + args := m.Called(ctx, origin) return args.Get(0).(int64), args.Error(1) } @@ -371,12 +373,9 @@ func TestRunAutoExchange_IdleRI_Skipped(t *testing.T) { // test for gap G10: before this fix, RunAutoExchange called // CancelAllPendingExchanges (store-wide), so the standalone ri_exchange_reshape // task would silently wipe out ladder-linked pending reshapes. After the fix, -// the standalone run calls CancelPendingExchangesByOrigin(ctx, false) which -// only cancels records where ladder_run_id IS NULL. -// -// Regression: pre-fix code routes through CancelAllPendingExchanges and would -// cancel with ladderScoped=nil (method absent). Post-fix: CancelPendingExchangesByOrigin -// is called with ladderScoped=false. +// the standalone run calls CancelPendingExchangesByOrigin with +// common.ExchangeOriginStandalone, which only cancels records where +// ladder_run_id IS NULL. func TestRunAutoExchange_StandaloneDoesNotCancelLadderPendings(t *testing.T) { t.Parallel() store := &mockExchangeStore{dailySpend: "0"} @@ -388,19 +387,19 @@ func TestRunAutoExchange_StandaloneDoesNotCancelLadderPendings(t *testing.T) { _, err := RunAutoExchange(context.Background(), params) require.NoError(t, err) - // Must have called the scoped cancel with ladderScoped=false, NOT the - // store-wide CancelAllPendingExchanges. cancelByOriginLastScoped tracks the + // Must have called the scoped cancel with the standalone origin, NOT the + // store-wide CancelAllPendingExchanges. cancelByOriginLast tracks the // argument passed to CancelPendingExchangesByOrigin. - require.NotNil(t, store.cancelByOriginLastScoped, + require.NotNil(t, store.cancelByOriginLast, "CancelPendingExchangesByOrigin must have been called (not the store-wide variant)") - assert.False(t, *store.cancelByOriginLastScoped, - "standalone run must cancel only non-ladder pendings (ladderScoped=false)") + assert.Equal(t, common.ExchangeOriginStandalone, *store.cancelByOriginLast, + "standalone run must cancel only non-ladder pendings (origin=standalone)") } // TestRunAutoExchange_LadderDoesNotCancelStandalonePendings verifies that a -// ladder-originated run calls CancelPendingExchangesByOrigin(ctx, true), which -// only cancels records where ladder_run_id IS NOT NULL, leaving standalone -// pending records intact. +// ladder-originated run calls CancelPendingExchangesByOrigin with +// common.ExchangeOriginLadder, which only cancels records where ladder_run_id +// IS NOT NULL, leaving standalone pending records intact. func TestRunAutoExchange_LadderDoesNotCancelStandalonePendings(t *testing.T) { t.Parallel() store := &mockExchangeStore{dailySpend: "0"} @@ -412,10 +411,10 @@ func TestRunAutoExchange_LadderDoesNotCancelStandalonePendings(t *testing.T) { _, err := RunAutoExchange(context.Background(), params) require.NoError(t, err) - require.NotNil(t, store.cancelByOriginLastScoped, + require.NotNil(t, store.cancelByOriginLast, "CancelPendingExchangesByOrigin must have been called") - assert.True(t, *store.cancelByOriginLastScoped, - "ladder run must cancel only ladder-linked pendings (ladderScoped=true)") + assert.Equal(t, common.ExchangeOriginLadder, *store.cancelByOriginLast, + "ladder run must cancel only ladder-linked pendings (origin=ladder)") } // ─── Dry-run tests ──────────────────────────────────────────────────────────── diff --git a/pkg/ladder/capability.go b/pkg/ladder/capability.go index 3bbf7b42b..fb004ab6b 100644 --- a/pkg/ladder/capability.go +++ b/pkg/ladder/capability.go @@ -81,6 +81,16 @@ type BufferReshapeConfig struct { LookbackDays int // DryRun, when true, simulates the reshape without executing any exchanges. DryRun bool + // LadderRunID links the exchange records this reshape creates to the ladder + // run that triggered it, so the exchange layer scopes its pending + // cancellation to this origin (ladder) instead of the standalone task's + // pendings (gap G10 / issue #1348). Nil means "no ladder run" (the caller is + // not a ladder run); the reshape runner then behaves as the standalone + // origin. The concrete runner (wired in L16) MUST forward this to + // exchange.RunAutoExchangeParams.LadderRunID — the exchangeRunner seam + // requires it so a future runner cannot silently drop it and reintroduce the + // cross-origin cancellation bug. + LadderRunID *string } // ReshapeSummary reports the outcome of a buffer reshape operation. diff --git a/providers/aws/ladder/interfaces.go b/providers/aws/ladder/interfaces.go index 67a0ed38b..e7db79a51 100644 --- a/providers/aws/ladder/interfaces.go +++ b/providers/aws/ladder/interfaces.go @@ -187,6 +187,16 @@ type spPurchaser interface { // AWSLadder only supplies the run configuration; injecting the full // exchange.RunAutoExchangeParams surface here would drag store and exchange // client dependencies into this package for no benefit. +// +// ladderRunID and dryRun are explicit parameters (not folded into +// RIExchangeConfig) so the seam is correct-by-construction: the concrete +// runner (wired in L16) is FORCED to forward them to +// exchange.RunAutoExchangeParams.{LadderRunID,DryRun}. A non-nil ladderRunID +// makes exchange.RunAutoExchange scope its pending-cancellation to the ladder +// origin (ladder_run_id IS NOT NULL); a nil ladderRunID keeps the standalone +// origin. Passing them here prevents a future runner from defaulting +// ladderRunID to nil and silently cancelling the standalone task's pendings +// (gap G10 / issue #1348). type exchangeRunner interface { - RunAutoExchange(ctx context.Context, cfg exchange.RIExchangeConfig) (*exchange.AutoExchangeResult, error) + RunAutoExchange(ctx context.Context, cfg exchange.RIExchangeConfig, ladderRunID *string, dryRun bool) (*exchange.AutoExchangeResult, error) } diff --git a/providers/aws/ladder/reshape.go b/providers/aws/ladder/reshape.go index 0e2e69a70..f9a5b27a6 100644 --- a/providers/aws/ladder/reshape.go +++ b/providers/aws/ladder/reshape.go @@ -28,20 +28,23 @@ const unlimitedCapUSD = math.MaxFloat64 // store, quote/execute client, offering lookup, and RI/utilization inventory // (the same wiring internal/server.executeRIExchangeReshape performs). // -// Origin-scoped cancellation: the underlying exchange.RunAutoExchange now -// calls CancelPendingExchangesByOrigin(ctx, true) — cancelling only -// ladder-linked pending records (ladder_run_id IS NOT NULL) — rather than the -// former store-wide CancelAllPendingExchanges. This means the standalone -// ri_exchange_reshape task's pending records are no longer affected by a ladder -// reshape run (gap G10, issue #1348). +// Origin-scoped cancellation (gap G10, issue #1348): exchange.RunAutoExchange +// scopes its pending-cancellation by origin — standalone runs +// (LadderRunID=nil) cancel only ladder_run_id IS NULL pendings, and ladder +// runs (LadderRunID set) cancel only ladder_run_id IS NOT NULL pendings. This +// method forwards cfg.LadderRunID through the exchangeRunner seam so a ladder +// reshape scopes correctly. NOTE: the standalone scoping is active today; the +// ladder-side scoping only takes effect once a concrete exchangeRunner is +// wired (L16) — no production caller wires the write side yet, so ReshapeBuffer +// fails loud with errWriteNotWired until then. The seam REQUIRES LadderRunID so +// the L16 runner cannot drop it and reintroduce the cross-origin bug. // -// DryRun is NOT yet wired end-to-end in this method: pkg/exchange now supports -// true dry-run (RunAutoExchangeParams.DryRun=true skips all mutations and -// returns Simulated outcomes), but passing cfg.DryRun through to the exchange -// runner requires the exchange runner parameter to be surfaced in the -// ReshapeRunner interface, which is tracked in L16. Until L16 lands, -// cfg.DryRun fails loud here; the engine previews reshapes via ActionReshape -// rationales without calling ReshapeBuffer. +// DryRun is forwarded through the seam (cfg.DryRun -> the runner's dryRun +// argument -> exchange.RunAutoExchangeParams.DryRun, which skips every +// mutation and returns Simulated outcomes). It is honored by the concrete +// runner wired in L16; today no production runner is wired, so a dry-run +// ReshapeBuffer call reaches only a test double or fails loud with +// errWriteNotWired. // // Config mapping (BufferReshapeConfig -> exchange.RIExchangeConfig): // @@ -82,20 +85,15 @@ func (a *AWSLadder) ReshapeBuffer(ctx context.Context, scope ladder.Scope, cfg l if err := a.validateScope(scope); err != nil { return ladder.ReshapeSummary{}, err } - if cfg.DryRun { - // Fail loud beats false simulation: the exchange flow's manual mode - // persists actionable approval records and cancels unrelated pendings - // store-wide — neither is a dry run. See the godoc warning above. - return ladder.ReshapeSummary{}, fmt.Errorf( - "ReshapeBuffer: dry-run is not supported by the AWS exchange flow yet (a true simulation mode in pkg/exchange is tracked upstream); the engine previews reshapes via ActionReshape rationales without calling ReshapeBuffer") - } - runCfg, err := buildRIExchangeConfig(cfg) if err != nil { return ladder.ReshapeSummary{}, fmt.Errorf("ReshapeBuffer: %w", err) } - result, err := a.exchange.RunAutoExchange(ctx, runCfg) + // Forward LadderRunID and DryRun through the seam so the concrete runner + // (L16) scopes cancellation to this origin and honors dry-run. See the + // godoc and the exchangeRunner interface for why these are explicit params. + result, err := a.exchange.RunAutoExchange(ctx, runCfg, cfg.LadderRunID, cfg.DryRun) if err != nil { return ladder.ReshapeSummary{}, fmt.Errorf("ReshapeBuffer: auto exchange run failed: %w", err) } @@ -126,9 +124,10 @@ func buildRIExchangeConfig(cfg ladder.BufferReshapeConfig) (exchange.RIExchangeC return exchange.RIExchangeConfig{}, fmt.Errorf("LookbackDays must be > 0, got %d", cfg.LookbackDays) } - // Mode is always auto: ReshapeBuffer rejects DryRun before this point - // (no true simulation mode exists in pkg/exchange; manual mode is not a - // simulation — see the ReshapeBuffer godoc warning). + // Mode is always auto. DryRun is not folded into RIExchangeConfig; it is + // forwarded as an explicit seam argument so the runner routes it to + // exchange.RunAutoExchangeParams.DryRun (which simulates without mutating, + // regardless of Mode). See the ReshapeBuffer godoc. return exchange.RIExchangeConfig{ Mode: string(exchange.ExchangeModeAuto), UtilizationThreshold: cfg.UtilizationThresholdPct, diff --git a/providers/aws/ladder/reshape_test.go b/providers/aws/ladder/reshape_test.go index 7430fa5b7..767a0e8ac 100644 --- a/providers/aws/ladder/reshape_test.go +++ b/providers/aws/ladder/reshape_test.go @@ -15,17 +15,22 @@ import ( ) // fakeExchangeRunner is a hermetic exchangeRunner double that records the -// config it received. Field order minimizes GC pointer-scan range. +// config and scoping arguments it received. Field order minimizes GC +// pointer-scan range. type fakeExchangeRunner struct { - err error - result *exchange.AutoExchangeResult - gotCfg exchange.RIExchangeConfig - calls int + err error + result *exchange.AutoExchangeResult + gotLadderRunID *string + gotCfg exchange.RIExchangeConfig + calls int + gotDryRun bool } -func (f *fakeExchangeRunner) RunAutoExchange(_ context.Context, cfg exchange.RIExchangeConfig) (*exchange.AutoExchangeResult, error) { +func (f *fakeExchangeRunner) RunAutoExchange(_ context.Context, cfg exchange.RIExchangeConfig, ladderRunID *string, dryRun bool) (*exchange.AutoExchangeResult, error) { f.calls++ f.gotCfg = cfg + f.gotLadderRunID = ladderRunID + f.gotDryRun = dryRun if f.result == nil && f.err == nil { return &exchange.AutoExchangeResult{Mode: cfg.Mode}, nil } @@ -137,20 +142,20 @@ func TestReshapeBuffer_ThresholdAndLookbackValidation(t *testing.T) { } } -func TestReshapeBuffer_DryRun_NotSupported_FailsLoudWithoutCallingRunner(t *testing.T) { - // DryRun must NOT be mapped onto the exchange flow's manual mode: manual - // mode persists pending ExchangeRecords with live approval tokens and - // RunAutoExchange unconditionally cancels ALL pending records store-wide - // first — neither is a simulation. The decided behavior is a loud error. +func TestReshapeBuffer_DryRun_ForwardedThroughSeam(t *testing.T) { + // pkg/exchange now supports a true dry-run (DryRun=true skips every mutation + // and returns Simulated outcomes), so ReshapeBuffer forwards cfg.DryRun + // through the seam instead of failing loud. The concrete runner (L16) honors + // it; here the fake asserts the flag reached the seam. ex := &fakeExchangeRunner{} a := newWiredLadder(t, &fakePurchaser{}, &fakePurchaser{}, ex) cfg := validReshapeCfg() cfg.DryRun = true _, err := a.ReshapeBuffer(context.Background(), testScope(), cfg) - require.Error(t, err) - assert.Contains(t, err.Error(), "not supported") - assert.Equal(t, 0, ex.calls, "a dry run must never reach the runner (it would cancel unrelated pending exchanges)") + require.NoError(t, err) + require.Equal(t, 1, ex.calls, "the runner must be reached so it can simulate") + assert.True(t, ex.gotDryRun, "cfg.DryRun must be forwarded to the runner's dryRun argument") } func TestReshapeBuffer_LiveRun_AlwaysAutoMode(t *testing.T) { @@ -163,6 +168,40 @@ func TestReshapeBuffer_LiveRun_AlwaysAutoMode(t *testing.T) { require.NoError(t, err) require.Equal(t, 1, ex.calls) assert.Equal(t, string(exchange.ExchangeModeAuto), ex.gotCfg.Mode) + assert.False(t, ex.gotDryRun, "a live run must forward dryRun=false") +} + +// TestReshapeBuffer_LadderRunID_ForwardedThroughSeam is the correct-by-construction +// regression for gap G10: a ladder-originated reshape MUST forward its +// LadderRunID through the exchangeRunner seam so exchange.RunAutoExchange scopes +// cancellation to the ladder origin. Before the seam carried LadderRunID, a +// future concrete runner would default it to nil (standalone scoping) and cancel +// the standalone task's pendings. +func TestReshapeBuffer_LadderRunID_ForwardedThroughSeam(t *testing.T) { + ex := &fakeExchangeRunner{} + a := newWiredLadder(t, &fakePurchaser{}, &fakePurchaser{}, ex) + + runID := "ladder-run-xyz" + cfg := validReshapeCfg() + cfg.LadderRunID = &runID + _, err := a.ReshapeBuffer(context.Background(), testScope(), cfg) + require.NoError(t, err) + require.Equal(t, 1, ex.calls) + require.NotNil(t, ex.gotLadderRunID, "LadderRunID must be forwarded, not dropped to nil") + assert.Equal(t, runID, *ex.gotLadderRunID, "the exact run ID must reach the runner for ladder-origin scoping") +} + +// TestReshapeBuffer_NilLadderRunID_ForwardedAsNil verifies a non-ladder caller +// forwards nil (standalone origin) rather than a fabricated value. +func TestReshapeBuffer_NilLadderRunID_ForwardedAsNil(t *testing.T) { + ex := &fakeExchangeRunner{} + a := newWiredLadder(t, &fakePurchaser{}, &fakePurchaser{}, ex) + + cfg := validReshapeCfg() // LadderRunID left nil + _, err := a.ReshapeBuffer(context.Background(), testScope(), cfg) + require.NoError(t, err) + require.Equal(t, 1, ex.calls) + assert.Nil(t, ex.gotLadderRunID, "a non-ladder reshape must forward nil (standalone origin)") } // --------------------------------------------------------------------------- @@ -257,7 +296,7 @@ func TestReshapeBuffer_NilResultWithoutError_IsContractViolation(t *testing.T) { // nilResultRunner deliberately violates the runner contract for the guard test. type nilResultRunner struct{} -func (n *nilResultRunner) RunAutoExchange(_ context.Context, _ exchange.RIExchangeConfig) (*exchange.AutoExchangeResult, error) { +func (n *nilResultRunner) RunAutoExchange(_ context.Context, _ exchange.RIExchangeConfig, _ *string, _ bool) (*exchange.AutoExchangeResult, error) { return nil, nil }