From 0bf718fd6c7192780012f32639a795a206f59c86 Mon Sep 17 00:00:00 2001 From: Cristian Magherusan-Stanciu Date: Thu, 8 Oct 2026 00:12:32 +0200 Subject: [PATCH] fix(exchange): atomically reserve RI exchange daily headroom Refs LeanerCloud/cloud-commitments-go#41 --- internal/analytics/collector_test.go | 3 + internal/config/interfaces.go | 1 + internal/config/store_postgres.go | 115 +++++++++++- .../config/store_postgres_pgxmock_test.go | 103 +++++++++++ ...i_exchange_reservation_integration_test.go | 168 ++++++++++++++++++ internal/mocks/stores.go | 6 + internal/server/test_helpers_test.go | 4 + 7 files changed, 392 insertions(+), 8 deletions(-) create mode 100644 internal/config/store_postgres_ri_exchange_reservation_integration_test.go diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index d7e912930..cd3e89fc1 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -330,6 +330,9 @@ func (m *mockConfigStore) ListStuckExecutions(ctx context.Context, statuses []st func (m *mockConfigStore) SaveRIExchangeRecord(ctx context.Context, record *config.RIExchangeRecord) error { return nil } +func (m *mockConfigStore) ReserveRIExchange(ctx context.Context, record *config.RIExchangeRecord, dailyCapUSD, perExchangeCapUSD string) (string, error) { + return "", errors.New("RI exchange reservation not configured in analytics test store") +} func (m *mockConfigStore) GetRIExchangeRecord(ctx context.Context, id string) (*config.RIExchangeRecord, error) { return nil, nil } diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 9d18edd62..009d662f5 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -362,6 +362,7 @@ type StoreInterface interface { // RI Exchange history SaveRIExchangeRecord(ctx context.Context, record *RIExchangeRecord) error + ReserveRIExchange(ctx context.Context, record *RIExchangeRecord, dailyCapUSD, perExchangeCapUSD string) (string, error) GetRIExchangeRecord(ctx context.Context, id string) (*RIExchangeRecord, error) GetRIExchangeRecordByToken(ctx context.Context, token string) (*RIExchangeRecord, error) GetRIExchangeHistory(ctx context.Context, since time.Time, limit int) ([]RIExchangeRecord, error) diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index 5bda5afe9..68f0bde0b 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -7,6 +7,8 @@ import ( "errors" "fmt" "hash/fnv" + "math/big" + "regexp" "sort" "strings" "time" @@ -63,6 +65,12 @@ var globalConfigLockKey = func() int64 { return int64(h.Sum64()) }() +var riExchangeDailySpendLockKey = func() int64 { + h := fnv.New64a() + _, _ = h.Write([]byte("cudly:ri_exchange:global_daily_spend")) + return int64(h.Sum64()) +}() + // GetGlobalConfig retrieves the global configuration. func (s *PostgresStore) GetGlobalConfig(ctx context.Context) (*GlobalConfig, error) { return getGlobalConfigFrom(ctx, s.db) @@ -2778,6 +2786,10 @@ func (s *PostgresStore) ClaimRIExchangeIdempotencyKey(ctx context.Context, key s // cloud-commitments-go), never a value previously read back from this // column, so hashing it here can never double-hash an already-hashed value. func (s *PostgresStore) SaveRIExchangeRecord(ctx context.Context, record *RIExchangeRecord) error { + return saveRIExchangeRecord(ctx, s.db, record) +} + +func saveRIExchangeRecord(ctx context.Context, q globalConfigExecutor, record *RIExchangeRecord) error { if record.ID == "" { record.ID = uuid.New().String() } @@ -2810,7 +2822,7 @@ func (s *PostgresStore) SaveRIExchangeRecord(ctx context.Context, record *RIExch ) 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, + _, err := q.Exec(ctx, query, record.ID, record.AccountID, record.ExchangeID, @@ -2841,6 +2853,91 @@ func (s *PostgresStore) SaveRIExchangeRecord(ctx context.Context, record *RIExch return nil } +// ReserveRIExchange atomically records the full execution ceiling before a provider call. +func (s *PostgresStore) ReserveRIExchange(ctx context.Context, record *RIExchangeRecord, dailyCapUSD, perExchangeCapUSD string) (string, error) { + initial, dailyCap, perCap, err := riExchangeReservationMicros(record, dailyCapUSD, perExchangeCapUSD) + if err != nil { + return "", err + } + + candidate := *record + var ceiling string + err = s.WithTx(ctx, func(tx pgx.Tx) error { + if _, lockErr := tx.Exec(ctx, "SELECT pg_advisory_xact_lock($1)", riExchangeDailySpendLockKey); lockErr != nil { + return fmt.Errorf("failed to lock RI exchange daily spend: %w", lockErr) + } + var spendText string + if scanErr := tx.QueryRow(ctx, riExchangeDailySpendQuery, time.Now().UTC()).Scan(&spendText); scanErr != nil { + return fmt.Errorf("failed to read RI exchange daily spend: %w", scanErr) + } + spend, parseErr := riExchangeMicros(spendText) + if parseErr != nil { + return fmt.Errorf("invalid RI exchange daily spend: %w", parseErr) + } + remaining := new(big.Int).Sub(dailyCap, spend) + if remaining.Cmp(perCap) > 0 { + remaining.Set(perCap) + } + if remaining.Sign() < 0 { + return fmt.Errorf("RI exchange daily cap exceeded: spend %s exceeds cap %s", spendText, dailyCapUSD) + } + if initial.Cmp(remaining) > 0 { + return fmt.Errorf("RI exchange daily cap exceeded: initial %s, remaining %s", record.PaymentDue, riExchangeUSD(remaining)) + } + ceiling = riExchangeUSD(remaining) + candidate.PaymentDue = ceiling + return saveRIExchangeRecord(ctx, tx, &candidate) + }) + if err != nil { + return "", err + } + *record = candidate + return ceiling, nil +} + +func riExchangeReservationMicros(record *RIExchangeRecord, dailyCapUSD, perExchangeCapUSD string) (initial, dailyCap, perCap *big.Int, err error) { + if record == nil || record.Status != "processing" || record.Mode != "auto" { + return nil, nil, nil, errors.New("auto RI exchange reservation requires a processing auto record") + } + initial, err = riExchangeMicros(record.PaymentDue) + if err != nil { + return nil, nil, nil, fmt.Errorf("invalid initial RI exchange payment: %w", err) + } + dailyCap, err = riExchangeMicros(dailyCapUSD) + if err != nil { + return nil, nil, nil, fmt.Errorf("invalid daily RI exchange cap: %w", err) + } + perCap, err = riExchangeMicros(perExchangeCapUSD) + if err != nil { + return nil, nil, nil, fmt.Errorf("invalid per-exchange RI cap: %w", err) + } + return initial, dailyCap, perCap, nil +} + +var riExchangeMicroScale = big.NewInt(1_000_000) +var riExchangeDecimalPattern = regexp.MustCompile(`^\d+(?:\.\d+)?$`) + +func riExchangeMicros(value string) (*big.Int, error) { + if !riExchangeDecimalPattern.MatchString(value) { + return nil, fmt.Errorf("invalid nonnegative decimal %q", value) + } + amount, ok := new(big.Rat).SetString(value) + if !ok { + return nil, fmt.Errorf("invalid nonnegative decimal %q", value) + } + amount.Mul(amount, new(big.Rat).SetInt(riExchangeMicroScale)) + if !amount.IsInt() || len(new(big.Int).Quo(amount.Num(), riExchangeMicroScale).String()) > 14 { + return nil, fmt.Errorf("decimal exceeds DECIMAL(20,6): %q", value) + } + return new(big.Int).Set(amount.Num()), nil +} + +func riExchangeUSD(micros *big.Int) string { + whole, fractional := new(big.Int), new(big.Int) + whole.QuoRem(micros, riExchangeMicroScale, fractional) + return fmt.Sprintf("%s.%06d", whole, fractional) +} + // GetRIExchangeRecord retrieves an RI exchange record by ID. func (s *PostgresStore) GetRIExchangeRecord(ctx context.Context, id string) (*RIExchangeRecord, error) { query := ` @@ -3057,13 +3154,7 @@ func (s *PostgresStore) FailRIExchange(ctx context.Context, id, errorMsg string) // it is updated_at (the moment the record transitioned to processing, i.e. // when it was approved). func (s *PostgresStore) GetRIExchangeDailySpend(ctx context.Context, date time.Time) (string, error) { - query := ` - SELECT COALESCE(SUM(payment_due), 0)::text - FROM ri_exchange_history - WHERE status IN ('completed', 'processing') - AND COALESCE(completed_at, updated_at) >= date_trunc('day', $1::timestamptz AT TIME ZONE 'UTC') - AND COALESCE(completed_at, updated_at) < date_trunc('day', $1::timestamptz AT TIME ZONE 'UTC') + INTERVAL '1 day' - ` + query := riExchangeDailySpendQuery var total string err := s.db.QueryRow(ctx, query, date).Scan(&total) @@ -3074,6 +3165,14 @@ func (s *PostgresStore) GetRIExchangeDailySpend(ctx context.Context, date time.T return total, nil } +const riExchangeDailySpendQuery = ` + SELECT COALESCE(SUM(payment_due), 0)::text + FROM ri_exchange_history + WHERE status IN ('completed', 'processing') + AND COALESCE(completed_at, updated_at) >= (date_trunc('day', $1::timestamptz AT TIME ZONE 'UTC') AT TIME ZONE 'UTC') + AND COALESCE(completed_at, updated_at) < (date_trunc('day', $1::timestamptz AT TIME ZONE 'UTC') + INTERVAL '1 day') AT TIME ZONE 'UTC' + ` + // CancelAllPendingExchanges cancels all pending RI exchange records regardless // of origin. Kept for interface compatibility; new callers should prefer // CancelPendingExchangesByOrigin to avoid cross-origin contamination. diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index 83eacf144..f1c215d17 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -2234,6 +2234,109 @@ func TestPGXMock_SaveRIExchangeRecord_InsertColumnAlignment(t *testing.T) { assert.NoError(t, mock.ExpectationsWereMet()) } +func TestPGXMock_ReserveRIExchange_CommitsCeilingBeforeReturn(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + mock.ExpectBegin() + mock.ExpectExec("SELECT pg_advisory_xact_lock").WithArgs(riExchangeDailySpendLockKey). + WillReturnResult(pgxmock.NewResult("SELECT", 1)) + mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg()). + WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("0.000001")) + mock.ExpectExec("INSERT INTO ri_exchange_history").WithArgs(anyArgsCfg(21)...). + WillReturnResult(pgxmock.NewResult("INSERT", 1)) + mock.ExpectCommit() + record := &RIExchangeRecord{AccountID: "account-a", Region: "us-east-1", PaymentDue: "0.000001", Status: "processing", Mode: "auto"} + ceiling, err := store.ReserveRIExchange(context.Background(), record, "1.000000", "1.000000") + require.NoError(t, err) + assert.Equal(t, "0.999999", ceiling) + assert.Equal(t, ceiling, record.PaymentDue) + assert.NotEmpty(t, record.ID) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestPGXMock_ReserveRIExchange_RollsBackWhenQuoteExceedsHeadroom(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + mock.ExpectBegin() + mock.ExpectExec("SELECT pg_advisory_xact_lock").WithArgs(riExchangeDailySpendLockKey). + WillReturnResult(pgxmock.NewResult("SELECT", 1)) + mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg()). + WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("101.000000")) + mock.ExpectRollback() + record := &RIExchangeRecord{PaymentDue: "900", Status: "processing", Mode: "auto"} + _, err := store.ReserveRIExchange(context.Background(), record, "1000", "1000") + require.ErrorContains(t, err, "daily cap exceeded") + assert.Empty(t, record.ID) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestPGXMock_ReserveRIExchange_DatabaseFailuresDoNotReturnCeiling(t *testing.T) { + for _, stage := range []string{"begin", "lock", "read", "insert"} { + t.Run(stage, func(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + dbErr := errors.New(stage + " failed") + if stage == "begin" { + mock.ExpectBegin().WillReturnError(dbErr) + } else { + mock.ExpectBegin() + lock := mock.ExpectExec("SELECT pg_advisory_xact_lock").WithArgs(riExchangeDailySpendLockKey) + if stage == "lock" { + lock.WillReturnError(dbErr) + } else { + lock.WillReturnResult(pgxmock.NewResult("SELECT", 1)) + read := mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg()) + if stage == "read" { + read.WillReturnError(dbErr) + } else { + read.WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("0")) + mock.ExpectExec("INSERT INTO ri_exchange_history").WithArgs(anyArgsCfg(21)...).WillReturnError(dbErr) + } + } + mock.ExpectRollback() + } + record := &RIExchangeRecord{PaymentDue: "900", Status: "processing", Mode: "auto"} + ceiling, err := store.ReserveRIExchange(context.Background(), record, "1000", "1000") + require.ErrorIs(t, err, dbErr) + assert.Empty(t, ceiling) + assert.Empty(t, record.ID) + assert.Equal(t, "900", record.PaymentDue) + assert.NoError(t, mock.ExpectationsWereMet()) + }) + } +} + +func TestPGXMock_ReserveRIExchange_CommitFailureDoesNotReturnCeiling(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + mock.ExpectBegin() + mock.ExpectExec("SELECT pg_advisory_xact_lock").WithArgs(riExchangeDailySpendLockKey). + WillReturnResult(pgxmock.NewResult("SELECT", 1)) + mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg()). + WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("0")) + mock.ExpectExec("INSERT INTO ri_exchange_history").WithArgs(anyArgsCfg(21)...). + WillReturnResult(pgxmock.NewResult("INSERT", 1)) + mock.ExpectCommit().WillReturnError(errors.New("commit failed")) + mock.ExpectRollback() + record := &RIExchangeRecord{PaymentDue: "900", Status: "processing", Mode: "auto"} + ceiling, err := store.ReserveRIExchange(context.Background(), record, "1000", "1000") + require.ErrorContains(t, err, "commit failed") + assert.Empty(t, ceiling) + assert.Empty(t, record.ID) + assert.Equal(t, "900", record.PaymentDue) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestRIExchangeMicros_RejectsInvalidPrecision(t *testing.T) { + for _, value := range []string{"", "-1", "1.0000001", "100000000000000", "1/2", "NaN"} { + _, err := riExchangeMicros(value) + require.Error(t, err, value) + } + amount, err := riExchangeMicros("0.000001") + require.NoError(t, err) + assert.Equal(t, "0.000001", riExchangeUSD(amount)) +} + // ─── CleanupOldExecutions ───────────────────────────────────────────────────── func TestPGXMock_CleanupOldExecutions_Success(t *testing.T) { diff --git a/internal/config/store_postgres_ri_exchange_reservation_integration_test.go b/internal/config/store_postgres_ri_exchange_reservation_integration_test.go new file mode 100644 index 000000000..b9e8bc076 --- /dev/null +++ b/internal/config/store_postgres_ri_exchange_reservation_integration_test.go @@ -0,0 +1,168 @@ +//go:build integration + +package config + +import ( + "context" + "math/big" + "net/url" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" +) + +func TestPostgresStoreDB_ReserveRIExchange_SerializesIndependentStores(t *testing.T) { + dsn := os.Getenv("RI_EXCHANGE_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("set RI_EXCHANGE_TEST_DATABASE_URL to a dedicated issue41 test database") + } + parsed, err := url.Parse(dsn) + require.NoError(t, err) + require.True(t, strings.HasPrefix(strings.TrimPrefix(parsed.Path, "/"), "issue41_"), "database must be dedicated to issue41 verification") + + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + newPool := func() (*pgxpool.Pool, error) { + cfg, parseErr := pgxpool.ParseConfig(dsn) + if parseErr != nil { + return nil, parseErr + } + cfg.AfterConnect = func(ctx context.Context, conn *pgx.Conn) error { + _, setErr := conn.Exec(ctx, "SET TIME ZONE 'America/Los_Angeles'") + return setErr + } + return pgxpool.NewWithConfig(ctx, cfg) + } + poolA, err := newPool() + require.NoError(t, err) + defer poolA.Close() + poolB, err := newPool() + require.NoError(t, err) + defer poolB.Close() + require.NoError(t, migrations.RunMigrations(ctx, poolA, getTestMigrationsPath(), "", "")) + storeA := &PostgresStore{db: poolA} + storeB := &PostgresStore{db: poolB} + beforeBoundaryRow, err := storeA.GetRIExchangeDailySpend(ctx, time.Now().UTC()) + require.NoError(t, err) + beforeMicros, err := riExchangeMicros(beforeBoundaryRow) + require.NoError(t, err) + utcOneAM := time.Now().UTC().Truncate(24 * time.Hour).Add(time.Hour) + boundaryRow := &RIExchangeRecord{AccountID: "444444444444", Region: "us-east-1", SourceRIIDs: []string{"ri-boundary"}, + SourceInstanceType: "m5.large", SourceCount: 1, TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: "1.000000", Status: "completed", Mode: "auto", CompletedAt: &utcOneAM} + require.NoError(t, storeA.SaveRIExchangeRecord(ctx, boundaryRow)) + baseline, err := storeA.GetRIExchangeDailySpend(ctx, time.Now().UTC()) + require.NoError(t, err) + baselineMicros, err := riExchangeMicros(baseline) + require.NoError(t, err) + require.Equal(t, 0, new(big.Int).Sub(baselineMicros, beforeMicros).Cmp(big.NewInt(1_000_000)), "UTC 01:00 exchange must count even when session timezone is Los Angeles") + cap := riExchangeUSD(new(big.Int).Add(baselineMicros, big.NewInt(1_000_000_000))) + + start := make(chan struct{}) + type reservation struct { + record *RIExchangeRecord + amount string + err error + } + results := make(chan reservation, 2) + var workers sync.WaitGroup + for i, store := range []*PostgresStore{storeA, storeB} { + workers.Add(1) + go func(i int, store *PostgresStore) { + defer workers.Done() + <-start + record := &RIExchangeRecord{ + AccountID: []string{"111111111111", "222222222222"}[i], Region: []string{"us-east-1", "eu-west-1"}[i], + SourceRIIDs: []string{"ri-" + string(rune('A'+i))}, SourceInstanceType: "m5.large", SourceCount: 1, + TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: "900.000000", Status: "processing", Mode: "auto", + } + amount, reserveErr := store.ReserveRIExchange(ctx, record, cap, "1000.000000") + results <- reservation{record: record, amount: amount, err: reserveErr} + }(i, store) + } + close(start) + workers.Wait() + close(results) + var winner *RIExchangeRecord + for result := range results { + if result.err == nil { + require.Nil(t, winner, "two stores reserved against the same daily headroom") + winner = result.record + require.Equal(t, "1000.000000", result.amount) + continue + } + require.ErrorContains(t, result.err, "daily cap exceeded") + } + require.NotNil(t, winner) + require.NotEmpty(t, winner.ID) + + var status, payment string + require.NoError(t, poolB.QueryRow(ctx, "SELECT status, payment_due::text FROM ri_exchange_history WHERE id = $1", winner.ID).Scan(&status, &payment)) + require.Equal(t, "processing", status) + require.Equal(t, "1000.000000", payment) + require.NoError(t, storeA.CompleteRIExchangeWithPayment(ctx, winner.ID, "exchange-accepted", "950.000000")) + require.NoError(t, poolB.QueryRow(ctx, "SELECT status, payment_due::text FROM ri_exchange_history WHERE id = $1", winner.ID).Scan(&status, &payment)) + require.Equal(t, "completed", status) + require.Equal(t, "950.000000", payment) + spend, err := storeB.GetRIExchangeDailySpend(ctx, time.Now().UTC()) + require.NoError(t, err) + require.Equal(t, riExchangeUSD(new(big.Int).Add(baselineMicros, big.NewInt(950_000_000))), spend) + + third := &RIExchangeRecord{AccountID: "333333333333", Region: "ap-south-1", SourceRIIDs: []string{"ri-C"}, + SourceInstanceType: "m5.large", SourceCount: 1, TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: "50.000000", Status: "processing", Mode: "auto"} + remaining, err := storeB.ReserveRIExchange(ctx, third, cap, "1000.000000") + require.NoError(t, err) + require.Equal(t, "50.000000", remaining) + + var rowsBefore int + require.NoError(t, poolA.QueryRow(ctx, "SELECT COUNT(*) FROM ri_exchange_history").Scan(&rowsBefore)) + lockTx, err := poolA.Begin(ctx) + require.NoError(t, err) + defer lockTx.Rollback(ctx) + _, err = lockTx.Exec(ctx, "SELECT pg_advisory_xact_lock($1)", riExchangeDailySpendLockKey) + require.NoError(t, err) + probe := &RIExchangeRecord{AccountID: "777777777777", Region: "us-west-2", SourceRIIDs: []string{"ri-lock-probe"}, + SourceInstanceType: "m5.large", SourceCount: 1, TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: "1.000000", Status: "processing", Mode: "auto"} + probeResult := make(chan error, 1) + go func() { + _, reserveErr := storeB.ReserveRIExchange(ctx, probe, riExchangeUSD(new(big.Int).Add(baselineMicros, big.NewInt(2_000_000_000))), "1.000000") + probeResult <- reserveErr + }() + observedWait := false + for deadline := time.Now().Add(3 * time.Second); time.Now().Before(deadline); { + var waiting bool + require.NoError(t, poolA.QueryRow(ctx, `SELECT EXISTS ( + SELECT 1 FROM pg_stat_activity WHERE wait_event_type = 'Lock' AND wait_event = 'advisory' + AND query LIKE 'SELECT pg_advisory_xact_lock%')`).Scan(&waiting)) + if waiting { + observedWait = true + break + } + select { + case reserveErr := <-probeResult: + t.Fatalf("reservation returned before lock release: %v", reserveErr) + case <-time.After(10 * time.Millisecond): + } + } + require.True(t, observedWait, "reservation must wait on the same global advisory lock") + var rowsDuring int + require.NoError(t, poolA.QueryRow(ctx, "SELECT COUNT(*) FROM ri_exchange_history").Scan(&rowsDuring)) + require.Equal(t, rowsBefore, rowsDuring) + require.NoError(t, lockTx.Commit(ctx)) + select { + case reserveErr := <-probeResult: + require.NoError(t, reserveErr) + case <-time.After(3 * time.Second): + t.Fatal("reservation did not proceed after global advisory lock release") + } +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 62a750721..a7caa7a01 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -660,6 +660,12 @@ func (m *MockConfigStore) SaveRIExchangeRecord(ctx context.Context, record *conf return args.Error(0) } +func (m *MockConfigStore) ReserveRIExchange(ctx context.Context, record *config.RIExchangeRecord, dailyCapUSD, perExchangeCapUSD string) (string, error) { + m.record("ReserveRIExchange", ctx, record, dailyCapUSD, perExchangeCapUSD) + args := m.Called(ctx, record, dailyCapUSD, perExchangeCapUSD) + return args.String(0), args.Error(1) +} + func (m *MockConfigStore) GetRIExchangeRecord(ctx context.Context, id string) (*config.RIExchangeRecord, error) { m.record("GetRIExchangeRecord", ctx, id) args := m.Called(ctx, id) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 2f297d7a1..5944314b5 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -2,6 +2,7 @@ package server import ( "context" + "errors" "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" @@ -180,6 +181,9 @@ func (m *mockConfigStoreForHealth) ListStuckExecutions(ctx context.Context, stat func (m *mockConfigStoreForHealth) SaveRIExchangeRecord(ctx context.Context, record *config.RIExchangeRecord) error { return nil } +func (m *mockConfigStoreForHealth) ReserveRIExchange(ctx context.Context, record *config.RIExchangeRecord, dailyCapUSD, perExchangeCapUSD string) (string, error) { + return "", errors.New("RI exchange reservation not configured in test store") +} func (m *mockConfigStoreForHealth) GetRIExchangeRecord(ctx context.Context, id string) (*config.RIExchangeRecord, error) { return nil, nil }