diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index 70e7359e8..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" @@ -320,6 +321,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, _ common.ExchangeOrigin) (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..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,6 +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 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, 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 81694fec6..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" @@ -2200,8 +2201,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 +2226,7 @@ func (s *PostgresStore) SaveRIExchangeRecord(ctx context.Context, record *RIExch record.CreatedAt, record.UpdatedAt, record.CreatedByUserID, + record.LadderRunID, ) if err != nil { @@ -2242,7 +2244,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 +2269,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 +2294,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 +2319,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 +2434,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 +2452,52 @@ func (s *PostgresStore) CancelAllPendingExchanges(ctx context.Context) (int64, e return result.RowsAffected(), nil } +// CancelPendingExchangesByOrigin cancels only pending records that match the +// 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 + switch origin { + case common.ExchangeOriginLadder: + query = ` + UPDATE ri_exchange_history + SET status = 'cancelled', updated_at = NOW() + WHERE status = 'pending' + AND ladder_run_id IS NOT NULL + ` + default: // common.ExchangeOriginStandalone (validated non-unknown above) + query = ` + UPDATE ri_exchange_history + SET status = 'cancelled', updated_at = NOW() + 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 %q: %w", origin, 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 +2506,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 +2527,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 +2551,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 +2569,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 +3291,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..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 @@ -165,6 +167,20 @@ 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() + + // 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, common.ExchangeOriginStandalone) + }) + + 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..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. @@ -988,6 +990,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 +1000,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 +1035,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 +1819,78 @@ func TestPGXMock_CancelAllPendingExchanges_Success(t *testing.T) { assert.Equal(t, int64(3), n) } +// ─── 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() + + // 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, common.ExchangeOriginStandalone) + require.NoError(t, err) + assert.Equal(t, int64(2), n) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestPGXMock_CancelPendingExchangesByOrigin_Ladder(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + ctx := context.Background() + + // 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, 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 ───────────────────────────────────────────────────── func TestPGXMock_CleanupOldExecutions_Success(t *testing.T) { diff --git a/internal/config/types.go b/internal/config/types.go index 90f252877..e895b872c 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -804,7 +804,20 @@ 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. + // + // 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"` CompletedAt *time.Time `json:"completed_at,omitempty"` diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index f1bcbfdd3..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,6 +527,15 @@ func (m *MockConfigStore) CancelAllPendingExchanges(ctx context.Context) (int64, return v, args.Error(1) } +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))) + } + 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..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,6 +295,10 @@ func (a *configExchangeStoreAdapter) CancelAllPendingExchanges(ctx context.Conte return a.store.CancelAllPendingExchanges(ctx) } +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) { cfgRecords, err := a.store.GetStaleProcessingExchanges(ctx, olderThan) if err != nil { @@ -335,6 +340,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 +365,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..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" @@ -790,12 +791,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, origin common.ExchangeOrigin) (int64, error) + getDailySpendFunc func(ctx context.Context, date time.Time) (string, error) } func (m *mockConfigStoreForExchange) GetGlobalConfig(ctx context.Context) (*config.GlobalConfig, error) { @@ -822,6 +824,13 @@ func (m *mockConfigStoreForExchange) CancelAllPendingExchanges(ctx context.Conte return 0, nil } +func (m *mockConfigStoreForExchange) CancelPendingExchangesByOrigin(ctx context.Context, origin common.ExchangeOrigin) (int64, error) { + if m.cancelPendingByOriginFunc != nil { + return m.cancelPendingByOriginFunc(ctx, origin) + } + 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..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" ) @@ -170,6 +171,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, _ common.ExchangeOrigin) (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/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 76238b3b0..b53be2f1f 100644 --- a/pkg/exchange/auto.go +++ b/pkg/exchange/auto.go @@ -29,16 +29,32 @@ 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 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. 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 @@ -77,6 +93,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 +138,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 +156,25 @@ 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 { + 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 (origin=%s)", cancelled, origin) + } } // 2. Log warning for stale processing records @@ -275,6 +316,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 +353,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 +372,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 +418,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 +478,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 +494,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 +534,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..babaaea6e 100644 --- a/pkg/exchange/auto_test.go +++ b/pkg/exchange/auto_test.go @@ -7,7 +7,10 @@ import ( "time" "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. @@ -17,6 +20,9 @@ type mockExchangeStore struct { staleRecords []ExchangeRecord dailySpend string dailySpendErr error + // cancelByOriginLast captures the origin argument of the last + // CancelPendingExchangesByOrigin call for assertion in scoping tests. + cancelByOriginLast *common.ExchangeOrigin } func (m *mockExchangeStore) SaveRIExchangeRecord(_ context.Context, record *ExchangeRecord) error { @@ -31,6 +37,11 @@ func (m *mockExchangeStore) CancelAllPendingExchanges(_ context.Context) (int64, return m.cancelledCount, nil } +func (m *mockExchangeStore) CancelPendingExchangesByOrigin(_ context.Context, origin common.ExchangeOrigin) (int64, error) { + m.cancelByOriginLast = &origin + return m.cancelledCount, nil +} + func (m *mockExchangeStore) GetStaleProcessingExchanges(_ context.Context, _ time.Duration) ([]ExchangeRecord, error) { return m.staleRecords, nil } @@ -50,6 +61,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, origin common.ExchangeOrigin) (int64, error) { + args := m.Called(ctx, origin) + 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 +366,148 @@ 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 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"} + 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 the standalone origin, NOT the + // store-wide CancelAllPendingExchanges. cancelByOriginLast tracks the + // argument passed to CancelPendingExchangesByOrigin. + require.NotNil(t, store.cancelByOriginLast, + "CancelPendingExchangesByOrigin must have been called (not the store-wide variant)") + 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 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"} + 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.cancelByOriginLast, + "CancelPendingExchangesByOrigin must have been called") + assert.Equal(t, common.ExchangeOriginLadder, *store.cancelByOriginLast, + "ladder run must cancel only ladder-linked pendings (origin=ladder)") +} + +// ─── 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/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 05d6947b7..f9a5b27a6 100644 --- a/providers/aws/ladder/reshape.go +++ b/providers/aws/ladder/reshape.go @@ -28,21 +28,23 @@ 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 (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 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 -// 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): // @@ -83,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) } @@ -127,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 }