diff --git a/internal/analytics/collector_test.go b/internal/analytics/collector_test.go index cd3e89fc1..edfa2805f 100644 --- a/internal/analytics/collector_test.go +++ b/internal/analytics/collector_test.go @@ -333,6 +333,10 @@ func (m *mockConfigStore) SaveRIExchangeRecord(ctx context.Context, record *conf 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) ReserveApprovedRIExchange(ctx context.Context, id, dailyCapUSD, perExchangeCapUSD string) (string, error) { + return "", errors.New("approved RI exchange reservation not configured in test store") +} func (m *mockConfigStore) GetRIExchangeRecord(ctx context.Context, id string) (*config.RIExchangeRecord, error) { return nil, nil } diff --git a/internal/api/exchange_helpers_test.go b/internal/api/exchange_helpers_test.go index 102ce7db0..d0bd87fc7 100644 --- a/internal/api/exchange_helpers_test.go +++ b/internal/api/exchange_helpers_test.go @@ -1,6 +1,6 @@ package api -// exchange_helpers_test.go — tests for checkDailyCap and federation IaC helpers. +// exchange_helpers_test.go — tests for federation IaC helpers. import ( "context" @@ -15,50 +15,6 @@ import ( // checkDailyCap // --------------------------------------------------------------------------- -func TestCheckDailyCap_WithinCap(t *testing.T) { - // $100 daily spend + $50 payment = $150, cap is $200 → allowed - reason := checkDailyCap("100.00", "50.00", 200.0) - assert.Equal(t, "", reason, "expected no reason when within cap") -} - -func TestCheckDailyCap_ExceedsCap(t *testing.T) { - // $150 spent + $100 payment = $250 > $200 cap → blocked - reason := checkDailyCap("150.00", "100.00", 200.0) - assert.NotEmpty(t, reason) - assert.Contains(t, reason, "daily cap exceeded") -} - -func TestCheckDailyCap_InvalidDailySpend(t *testing.T) { - // Unparseable daily spend → fail-safe block - reason := checkDailyCap("not-a-number", "50.00", 200.0) - assert.NotEmpty(t, reason) - assert.Contains(t, reason, "daily spend check failed") -} - -func TestCheckDailyCap_InvalidPaymentDue(t *testing.T) { - // H1 fix: an unparseable payment-due string must fail closed (return a - // blocking reason) instead of being treated as $0. Proceeding as $0 would - // allow an exchange of unknown cost through the daily cap check. - reason := checkDailyCap("100.00", "not-a-number", 500.0) - assert.NotEmpty(t, reason, "unparseable payment due must block the exchange (fail closed)") - assert.Contains(t, reason, "could not parse payment due") -} - -func TestCheckDailyCap_ExactlyAtCap(t *testing.T) { - // $100 spent + $100 payment = $200 == $200 cap → allowed (not strictly greater) - reason := checkDailyCap("100.00", "100.00", 200.0) - assert.Equal(t, "", reason) -} - -func TestCheckDailyCap_ZeroSpend(t *testing.T) { - reason := checkDailyCap("0.00", "50.00", 200.0) - assert.Equal(t, "", reason) -} - -// --------------------------------------------------------------------------- -// getFederationIaC — accessible validation paths (no DB required) -// --------------------------------------------------------------------------- - func TestHandler_getFederationIaC_MissingTarget(t *testing.T) { ctx := context.Background() h := federationHandler() diff --git a/internal/api/handler_ri_exchange.go b/internal/api/handler_ri_exchange.go index 80110685f..2d601f6b6 100644 --- a/internal/api/handler_ri_exchange.go +++ b/internal/api/handler_ri_exchange.go @@ -2593,17 +2593,6 @@ func (h *Handler) retryCompleteWithPayment(ctx context.Context, id, exchangeID, return err } -// handlerChooseEffectiveCap returns the smaller of perExchangeCap and daily -// headroom (dailyCap - dailySpent), bounding Execute's MaxPaymentDueUSD so a -// fresh re-quote cannot exceed the remaining daily budget (H2 fix). -func handlerChooseEffectiveCap(dailyCap, dailySpent, perExchangeCap *big.Rat) *big.Rat { - remaining := new(big.Rat).Sub(dailyCap, dailySpent) - if remaining.Cmp(perExchangeCap) < 0 { - return remaining - } - return perExchangeCap -} - // handlerAcceptedAmount extracts the confirmed payment amount from a fresh // Execute quote, falling back to fallback when freshQ is nil or empty (H3 fix). // @@ -2620,39 +2609,26 @@ func handlerAcceptedAmount(freshQ *exchange.ExchangeQuoteSummary, fallback strin return fallback } -// checkCapsAndComputeHeadroom validates the spending-cap configuration, runs the -// daily-cap check, and computes the effective MaxPaymentDueUSD that Execute must -// not exceed (H2: remaining daily headroom vs per-exchange cap, whichever is -// smaller). Returns a non-empty reason string on any failure so the caller can -// forward it to failExchange. -func checkCapsAndComputeHeadroom(dailySpendStr, paymentDue string, cfg *config.GlobalConfig) (effectiveCap *big.Rat, reason string) { +// exchangeCapStrings validates the spending-cap configuration and returns the +// daily and per-exchange caps as decimal strings for the atomic reservation. +// A non-empty reason means the caller must forward it to failExchange. +func exchangeCapStrings(cfg *config.GlobalConfig) (daily, perExchange, reason string) { if cfg.RIExchangeMaxDailyUSD == 0 { - return nil, "daily spending cap is not configured (RIExchangeMaxDailyUSD is 0)" - } - if reason := checkDailyCap(dailySpendStr, paymentDue, cfg.RIExchangeMaxDailyUSD); reason != "" { - return nil, reason + return "", "", "daily spending cap is not configured (RIExchangeMaxDailyUSD is 0)" } if cfg.RIExchangeMaxPerExchangeUSD == 0 { - return nil, "per-exchange spending cap is not configured (RIExchangeMaxPerExchangeUSD is 0)" - } - // checkDailyCap already verified dailySpendStr is parseable; a second failure - // is an internal error - fail closed to avoid executing with wrong headroom. - dailySpent, err := exchange.ParseDecimalRat(dailySpendStr) - if err != nil || dailySpent == nil { - return nil, fmt.Sprintf("daily spend re-parse failed (internal error): %v", err) + return "", "", "per-exchange spending cap is not configured (RIExchangeMaxPerExchangeUSD is 0)" } - dailyCap := new(big.Rat).SetFloat64(cfg.RIExchangeMaxDailyUSD) - perExchangeCap := new(big.Rat).SetFloat64(cfg.RIExchangeMaxPerExchangeUSD) - return handlerChooseEffectiveCap(dailyCap, dailySpent, perExchangeCap), "" + daily = new(big.Rat).SetFloat64(cfg.RIExchangeMaxDailyUSD).FloatString(6) + perExchange = new(big.Rat).SetFloat64(cfg.RIExchangeMaxPerExchangeUSD).FloatString(6) + return daily, perExchange, "" } -// executeApprovedExchange checks caps and executes the exchange after approval. +// executeApprovedExchange reserves the execution ceiling atomically under the +// shared daily-cap lock, then executes the exchange within that ceiling. The +// ceiling is the full permitted amount (min of per-exchange cap and daily +// headroom), so a fresh re-quote cannot push the day over the cap. func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record *config.RIExchangeRecord) (any, error) { - dailySpendStr, err := h.config.GetRIExchangeDailySpend(ctx, time.Now()) - if err != nil { - return h.failExchange(ctx, id, "daily spending cap check failed") - } - globalCfg, err := h.config.GetGlobalConfig(ctx) if err != nil { return h.failExchange(ctx, id, "config load failed") @@ -2666,10 +2642,22 @@ func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record return h.failExchange(ctx, id, "exchange record has no region; cannot execute safely") } - effectiveCap, reason := checkCapsAndComputeHeadroom(dailySpendStr, record.PaymentDue, globalCfg) + dailyCapUSD, perExchangeCapUSD, reason := exchangeCapStrings(globalCfg) if reason != "" { return h.failExchange(ctx, id, reason) } + ceiling, err := h.config.ReserveApprovedRIExchange(ctx, id, dailyCapUSD, perExchangeCapUSD) + if err != nil { + logging.Warnf("RI exchange %s reservation refused: %v", id, err) + if errors.Is(err, config.ErrRIExchangeDailyCapExceeded) { + return h.failExchange(ctx, id, err.Error()) + } + return h.failExchange(ctx, id, "daily spending cap check failed") + } + effectiveCap, ok := new(big.Rat).SetString(ceiling) + if !ok { + return h.failExchange(ctx, id, fmt.Sprintf("reserved ceiling %q is not a decimal", ceiling)) + } execFn := exchange.ExecuteExchange if h.executeExchangeFn != nil { @@ -2686,9 +2674,8 @@ func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record return h.failExchange(ctx, id, execErr.Error()) } - // H3: persist the amount AWS actually accepted, not the stale pre-execution - // quote stored in record.PaymentDue. - acceptedPaymentDue := handlerAcceptedAmount(freshQ, record.PaymentDue) + // H3: persist the amount AWS actually accepted, not the reserved ceiling. + acceptedPaymentDue := handlerAcceptedAmount(freshQ, ceiling) // H4: retry the ledger write via retryCompleteWithPayment; persistent // failure is returned as an error (HTTP 500) so the caller knows money @@ -2703,35 +2690,6 @@ func (h *Handler) executeApprovedExchange(ctx context.Context, id string, record return map[string]any{"status": "completed", "exchange_id": exchangeID}, nil } -// checkDailyCap verifies the exchange payment won't exceed the daily spending cap. -// Returns an empty string if within cap, or a reason string if exceeded or if -// either input cannot be parsed (fail closed on parse errors). -func checkDailyCap(dailySpendStr, paymentDueStr string, maxDailyUSD float64) string { - dailyCap := new(big.Rat).SetFloat64(maxDailyUSD) - dailySpent, err := exchange.ParseDecimalRat(dailySpendStr) - if err != nil || dailySpent == nil { - // A parse failure means we cannot determine today's spend; treat as a cap - // check failure to avoid under-counting spend (fail-safe). - logging.Warnf("checkDailyCap: failed to parse daily spend string %q: %v; blocking exchange to avoid exceeding cap", dailySpendStr, err) - return fmt.Sprintf("daily spend check failed: could not parse today's spend value %q", dailySpendStr) - } - paymentDue, err := exchange.ParseDecimalRat(paymentDueStr) - if err != nil || paymentDue == nil { - // H1 fix: fail closed on an unparseable payment-due string instead of - // treating it as $0. An unparseable value means we cannot determine the - // true cost of this exchange, so proceeding risks exceeding the cap. - logging.Warnf("checkDailyCap: failed to parse payment due string %q: %v; blocking exchange to avoid cap bypass", paymentDueStr, err) - return fmt.Sprintf("daily spend check failed: could not parse payment due value %q", paymentDueStr) - } - - newTotal := new(big.Rat).Add(dailySpent, paymentDue) - if newTotal.Cmp(dailyCap) > 0 { - return fmt.Sprintf("daily cap exceeded: spent $%s + payment $%s > cap $%.2f", - dailySpent.FloatString(2), paymentDue.FloatString(2), maxDailyUSD) - } - return "" -} - // rejectRIExchange handles rejection of a pending RI exchange via token. func (h *Handler) rejectRIExchange(ctx context.Context, id, token string) (any, error) { if err := validateUUID(id); err != nil { diff --git a/internal/api/handler_ri_exchange_test.go b/internal/api/handler_ri_exchange_test.go index 9afaea356..188c13e81 100644 --- a/internal/api/handler_ri_exchange_test.go +++ b/internal/api/handler_ri_exchange_test.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "math/big" + "strings" "testing" "time" @@ -392,8 +393,8 @@ func TestApproveRIExchange_SessionAdmin(t *testing.T) { mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-123"}, PaymentDue: "100.00"}, nil) - // executeApprovedExchange calls GetRIExchangeDailySpend + GetGlobalConfig - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + // executeApprovedExchange calls GetGlobalConfig + ReserveApprovedRIExchange + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -453,7 +454,7 @@ func TestApproveRIExchange_SessionApproveOwn(t *testing.T) { mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-1"}, PaymentDue: "50.00"}, nil) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -534,7 +535,7 @@ func TestApproveRIExchange_SessionScope(t *testing.T) { }, nil).Maybe() mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-1"}, PaymentDue: "50.00"}, nil).Maybe() - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil).Maybe() + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil).Maybe() mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil).Maybe() mockStore.On("StampRIExchangeApprovedBy", ctx, id, mock.Anything).Return(nil).Maybe() @@ -627,7 +628,7 @@ func fourEyesRIExchangeFixture(t *testing.T, creatorID *string, approverID strin }, nil) mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-1"}, PaymentDue: "50.00"}, nil) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("FailRIExchange", ctx, id, mock.AnythingOfType("string")).Return(nil) mockStore.On("StampRIExchangeApprovedBy", ctx, id, mock.Anything).Return(nil) @@ -768,7 +769,7 @@ func TestApproveRIExchange_TokenPathSkipsStampWhenExecutionErrors(t *testing.T) RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", &approver). Return(&config.RIExchangeRecord{ID: id, Status: "processing"}, nil) mockStore.On("CompleteRIExchangeWithPayment", ctx, id, "exch-518", "0"). @@ -807,7 +808,7 @@ func TestApproveRIExchange_LegacyTokenStillWorks(t *testing.T) { }, nil) mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing"}, nil) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -2238,7 +2239,6 @@ func TestExecuteApprovedExchange_EmptyRecordRegionFails(t *testing.T) { t.Cleanup(func() { mockStore.AssertExpectations(t) }) h := &Handler{config: mockStore} - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -2263,6 +2263,7 @@ func TestExecuteApprovedExchange_EmptyRecordRegionFails(t *testing.T) { require.True(t, ok) assert.Equal(t, "failed", respMap["status"]) assert.Contains(t, respMap["reason"], "no region") + mockStore.AssertNotCalled(t, "ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything) } // TestClassifyRecsAge pins the staleness classification thresholds for @@ -2327,7 +2328,7 @@ func TestApproveRIExchange_SessionActorStamped(t *testing.T) { mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.MatchedBy(func(a *string) bool { return a != nil && *a == actorID }), ).Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-1"}, PaymentDue: "10.00"}, nil) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil).Maybe() mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, }, nil) @@ -2366,42 +2367,6 @@ func TestRejectRIExchange_TokenPathActorIsNil(t *testing.T) { require.NoError(t, err) } -// ─── H1: checkDailyCap fails closed on unparseable paymentDue ───────────────── - -// TestCheckDailyCap_UnparseablePaymentDue_FailsClosed is the regression test -// for H1: checkDailyCap must return a blocking reason when paymentDueStr cannot -// be parsed, never treat it as $0 (which would allow an unknown-cost exchange -// to proceed when the daily cap might be exceeded). -func TestCheckDailyCap_UnparseablePaymentDue_FailsClosed(t *testing.T) { - t.Parallel() - cases := []struct { - name string - paymentDueStr string - }{ - {"not-a-number", "not-a-number"}, - {"empty string", ""}, - } - for _, tc := range cases { - tc := tc - t.Run(tc.name, func(t *testing.T) { - t.Parallel() - reason := checkDailyCap("100.00", tc.paymentDueStr, 500.0) - assert.NotEmpty(t, reason, - "checkDailyCap must fail closed when paymentDueStr=%q cannot be parsed", tc.paymentDueStr) - assert.Contains(t, reason, "could not parse payment due", - "reason must reference the payment-due parsing failure, not treat it as $0") - }) - } -} - -// TestCheckDailyCap_ValidPaymentDue_WithinCap_Passes verifies the happy path: -// valid parseable amounts within the daily cap must return an empty reason. -func TestCheckDailyCap_ValidPaymentDue_WithinCap_Passes(t *testing.T) { - t.Parallel() - reason := checkDailyCap("100.00", "50.00", 500.0) - assert.Empty(t, reason, "within-cap exchange must not be blocked") -} - // ─── H2: effectiveCap bounded by daily headroom ─────────────────────────────── // TestExecuteApprovedExchange_EffectiveCap_BoundedByDailyHeadroom is the @@ -2416,8 +2381,9 @@ func TestExecuteApprovedExchange_EffectiveCap_BoundedByDailyHeadroom(t *testing. mockStore := new(MockConfigStore) t.Cleanup(func() { mockStore.AssertExpectations(t) }) - // dailySpent=$450, dailyCap=$500, perExchangeCap=$100 -> headroom=$50 - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("450.00", nil) + // dailySpent=$450, dailyCap=$500, perExchangeCap=$100 -> the store reserves + // the $50 headroom as the ceiling. + mockStore.On("ReserveApprovedRIExchange", ctx, id, "500.000000", "100.000000").Return("50.000000", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 500, RIExchangeMaxPerExchangeUSD: 100, @@ -2468,7 +2434,7 @@ func TestExecuteApprovedExchange_AcceptedAmountFromFreshQuote(t *testing.T) { mockStore := new(MockConfigStore) t.Cleanup(func() { mockStore.AssertExpectations(t) }) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -2519,7 +2485,7 @@ func TestExecuteApprovedExchange_LedgerWriteFailure_ReturnsError(t *testing.T) { mockStore := new(MockConfigStore) t.Cleanup(func() { mockStore.AssertExpectations(t) }) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -2618,3 +2584,91 @@ func TestExecuteExchange_FourEyesPolicyLookupErrorFailsClosed(t *testing.T) { _, isClient := IsClientError(err) assert.False(t, isClient, "a lookup failure is a server error, not a client refusal") } + +// TestExecuteApprovedExchange_ReservationRefused_NoExecute asserts that when the +// atomic reservation refuses (daily headroom gone), the exchange fails and the +// provider is never called (#656). +func TestExecuteApprovedExchange_ReservationRefused_NoExecute(t *testing.T) { + ctx := context.Background() + const id = "550e8400-e29b-41d4-a716-000000000656" + + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ + RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, + }, nil) + mockStore.On("ReserveApprovedRIExchange", ctx, id, "1000.000000", "500.000000"). + Return("", fmt.Errorf("%w: initial 600.000000, remaining 400.000000", config.ErrRIExchangeDailyCapExceeded)) + mockStore.On("FailRIExchange", ctx, id, mock.MatchedBy(func(r string) bool { + return strings.Contains(r, "daily cap exceeded") + })).Return(nil) + + called := false + h := &Handler{config: mockStore, executeExchangeFn: func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + called = true + return "", nil, nil + }} + resp, err := h.executeApprovedExchange(ctx, id, &config.RIExchangeRecord{ + ID: id, Region: "us-east-1", SourceRIIDs: []string{"ri-1"}, TargetOfferingID: "o", TargetCount: 1, PaymentDue: "600.00", + }) + require.NoError(t, err) + assert.Equal(t, "failed", resp.(map[string]any)["status"]) + assert.False(t, called, "provider must not be called when the reservation is refused") +} + +// TestExecuteApprovedExchange_SettlesAcceptedAmountWithinReservedCeiling asserts +// the full reserved ceiling (not the initial quote) bounds Execute, and the fresh +// accepted amount is what gets settled on the same row (#656). +func TestExecuteApprovedExchange_SettlesAcceptedAmountWithinReservedCeiling(t *testing.T) { + ctx := context.Background() + const id = "550e8400-e29b-41d4-a716-000000000657" + + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ + RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, + }, nil) + mockStore.On("ReserveApprovedRIExchange", ctx, id, "1000.000000", "500.000000").Return("400.000000", nil) + mockStore.On("CompleteRIExchangeWithPayment", ctx, id, "exch-1", "390.000000").Return(nil) + + var got exchange.ExchangeExecuteRequest + h := &Handler{config: mockStore, executeExchangeFn: func(_ context.Context, req exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + got = req + return "exch-1", &exchange.ExchangeQuoteSummary{PaymentDueUSDStr: "390.000000"}, nil + }} + _, err := h.executeApprovedExchange(ctx, id, &config.RIExchangeRecord{ + ID: id, Region: "us-east-1", SourceRIIDs: []string{"ri-1"}, TargetOfferingID: "o", TargetCount: 1, PaymentDue: "100.00", + }) + require.NoError(t, err) + require.NotNil(t, got.MaxPaymentDueUSD) + assert.Equal(t, 0, got.MaxPaymentDueUSD.Cmp(big.NewRat(400, 1)), "ceiling must be the reserved $400, not the $100 initial quote") +} + +// TestExecuteApprovedExchange_ReservationInfrastructureError_HidesDetail asserts +// that a non-cap reservation error (lock/DB failure) is logged, not echoed to the +// approver, and the provider is never called. +func TestExecuteApprovedExchange_ReservationInfrastructureError_HidesDetail(t *testing.T) { + ctx := context.Background() + const id = "550e8400-e29b-41d4-a716-000000000658" + + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ + RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, + }, nil) + mockStore.On("ReserveApprovedRIExchange", ctx, id, "1000.000000", "500.000000"). + Return("", errors.New("failed to lock RI exchange daily spend: connection reset by peer")) + mockStore.On("FailRIExchange", ctx, id, "daily spending cap check failed").Return(nil) + + called := false + h := &Handler{config: mockStore, executeExchangeFn: func(context.Context, exchange.ExchangeExecuteRequest) (string, *exchange.ExchangeQuoteSummary, error) { + called = true + return "", nil, nil + }} + resp, err := h.executeApprovedExchange(ctx, id, &config.RIExchangeRecord{ + ID: id, Region: "us-east-1", SourceRIIDs: []string{"ri-1"}, TargetOfferingID: "o", TargetCount: 1, PaymentDue: "10.00", + }) + require.NoError(t, err) + assert.Equal(t, "daily spending cap check failed", resp.(map[string]any)["reason"]) + assert.False(t, called) +} diff --git a/internal/api/middleware_test.go b/internal/api/middleware_test.go index 20baaed84..6b046d4a5 100644 --- a/internal/api/middleware_test.go +++ b/internal/api/middleware_test.go @@ -517,7 +517,7 @@ func TestApproveRIExchangeViaSession_RequiresCSRF(t *testing.T) { }, nil).Maybe() mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-123"}, PaymentDue: "100.00"}, nil).Maybe() - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil).Maybe() + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil).Maybe() mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -593,7 +593,7 @@ func TestApproveRIExchange_CSRFFailureDoesNotFallThroughToToken(t *testing.T) { }, nil).Maybe() mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-123"}, PaymentDue: "100.00"}, nil).Maybe() - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil).Maybe() + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil).Maybe() mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, @@ -649,7 +649,7 @@ func TestApproveRIExchangeViaSession_PassesCSRF(t *testing.T) { }, nil).Once() mockStore.On("TransitionRIExchangeStatus", ctx, id, "pending", "processing", mock.Anything). Return(&config.RIExchangeRecord{ID: id, Status: "processing", SourceRIIDs: []string{"ri-123"}, PaymentDue: "100.00"}, nil) - mockStore.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil) + mockStore.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil) mockStore.On("GetGlobalConfig", ctx).Return(&config.GlobalConfig{ RIExchangeMaxDailyUSD: 1000, RIExchangeMaxPerExchangeUSD: 500, diff --git a/internal/api/ri_exchange_approve_reject_status_test.go b/internal/api/ri_exchange_approve_reject_status_test.go index 8ef3ec3a6..cadd2b1fb 100644 --- a/internal/api/ri_exchange_approve_reject_status_test.go +++ b/internal/api/ri_exchange_approve_reject_status_test.go @@ -69,7 +69,7 @@ func riStatusHandler(rec riStatusRecord, o riStatusOpts) *Handler { } store.On("TransitionRIExchangeStatus", mock.Anything, riStatusID, "pending", mock.Anything, mock.Anything). Return(transitioned, nil).Maybe() - store.On("GetRIExchangeDailySpend", mock.Anything, mock.Anything).Return("0", nil).Maybe() + store.On("ReserveApprovedRIExchange", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return("500.000000", nil).Maybe() store.On("GetGlobalConfig", mock.Anything).Return(&config.GlobalConfig{RequireDifferentApprover: o.fourEyes}, nil).Maybe() store.On("FailRIExchange", mock.Anything, riStatusID, mock.Anything).Return(nil).Maybe() store.On("StampRIExchangeApprovedBy", mock.Anything, riStatusID, mock.Anything).Return(nil).Maybe() diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 009d662f5..4043bf627 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -363,6 +363,10 @@ type StoreInterface interface { // RI Exchange history SaveRIExchangeRecord(ctx context.Context, record *RIExchangeRecord) error ReserveRIExchange(ctx context.Context, record *RIExchangeRecord, dailyCapUSD, perExchangeCapUSD string) (string, error) + // ReserveApprovedRIExchange reserves the execution ceiling for an already + // processing (approved) exchange under the shared daily-cap lock, excluding + // the row's own amount, and stores the ceiling as its payment_due. + ReserveApprovedRIExchange(ctx context.Context, id, 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 68f0bde0b..085c8be24 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -2863,38 +2863,104 @@ func (s *PostgresStore) ReserveRIExchange(ctx context.Context, record *RIExchang 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 reserveErr error + ceiling, reserveErr = reserveRIExchangeCeiling(ctx, tx, nil, record.PaymentDue, initial, dailyCap, perCap, dailyCapUSD) + if reserveErr != nil { + return reserveErr } - 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) + candidate.PaymentDue = ceiling + return saveRIExchangeRecord(ctx, tx, &candidate) + }) + if err != nil { + return "", err + } + *record = candidate + return ceiling, nil +} + +// ReserveApprovedRIExchange reserves the full execution ceiling for an approved +// exchange whose row is already processing. It runs under the same global +// advisory lock as ReserveRIExchange, so manual approvals and AUTO exchanges +// share one headroom protocol, and it leaves the row's own amount out of the +// spend it checks against. The row's payment_due becomes the reserved ceiling +// until CompleteRIExchangeWithPayment settles the accepted amount. +func (s *PostgresStore) ReserveApprovedRIExchange(ctx context.Context, id, dailyCapUSD, perExchangeCapUSD string) (string, error) { + dailyCap, err := riExchangeMicros(dailyCapUSD) + if err != nil { + return "", fmt.Errorf("invalid daily RI exchange cap: %w", err) + } + perCap, err := riExchangeMicros(perExchangeCapUSD) + if err != nil { + return "", fmt.Errorf("invalid per-exchange RI cap: %w", err) + } + + var ceiling string + err = s.WithTx(ctx, func(tx pgx.Tx) error { + var initialText string + scanErr := tx.QueryRow(ctx, + `SELECT payment_due::text FROM ri_exchange_history WHERE id = $1 AND status = 'processing' FOR UPDATE`, id, + ).Scan(&initialText) + if errors.Is(scanErr, pgx.ErrNoRows) { + return fmt.Errorf("ri exchange %s is not processing", id) } - spend, parseErr := riExchangeMicros(spendText) - if parseErr != nil { - return fmt.Errorf("invalid RI exchange daily spend: %w", parseErr) + if scanErr != nil { + return fmt.Errorf("failed to read RI exchange %s: %w", id, scanErr) } - remaining := new(big.Int).Sub(dailyCap, spend) - if remaining.Cmp(perCap) > 0 { - remaining.Set(perCap) + initial, parseErr := riExchangeMicros(initialText) + if parseErr != nil { + return fmt.Errorf("invalid initial RI exchange payment: %w", parseErr) } - if remaining.Sign() < 0 { - return fmt.Errorf("RI exchange daily cap exceeded: spend %s exceeds cap %s", spendText, dailyCapUSD) + var reserveErr error + ceiling, reserveErr = reserveRIExchangeCeiling(ctx, tx, &id, initialText, initial, dailyCap, perCap, dailyCapUSD) + if reserveErr != nil { + return reserveErr } - if initial.Cmp(remaining) > 0 { - return fmt.Errorf("RI exchange daily cap exceeded: initial %s, remaining %s", record.PaymentDue, riExchangeUSD(remaining)) + _, updErr := tx.Exec(ctx, + `UPDATE ri_exchange_history SET payment_due = $2::numeric, updated_at = NOW() WHERE id = $1`, id, ceiling) + if updErr != nil { + return fmt.Errorf("failed to record RI exchange reservation: %w", updErr) } - ceiling = riExchangeUSD(remaining) - candidate.PaymentDue = ceiling - return saveRIExchangeRecord(ctx, tx, &candidate) + return nil }) if err != nil { return "", err } - *record = candidate return ceiling, nil } +// ErrRIExchangeDailyCapExceeded marks a reservation refused because the +// initial quote does not fit the remaining daily or per-exchange headroom. +var ErrRIExchangeDailyCapExceeded = errors.New("RI exchange daily cap exceeded") + +// reserveRIExchangeCeiling takes the global daily-spend lock (held until tx ends) +// and returns min(perExchangeCap, dailyCap - spend) as the execution ceiling, +// refusing when the initial quote does not fit. excludeID leaves one row's own +// amount out of spend so an approval does not count itself twice. +func reserveRIExchangeCeiling(ctx context.Context, tx pgx.Tx, excludeID *string, initialText string, initial, dailyCap, perCap *big.Int, dailyCapUSD string) (string, 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(), excludeID).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("%w: spend %s exceeds cap %s", ErrRIExchangeDailyCapExceeded, spendText, dailyCapUSD) + } + if initial.Cmp(remaining) > 0 { + return "", fmt.Errorf("%w: initial %s, remaining %s", ErrRIExchangeDailyCapExceeded, initialText, riExchangeUSD(remaining)) + } + return riExchangeUSD(remaining), 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") @@ -3157,7 +3223,7 @@ func (s *PostgresStore) GetRIExchangeDailySpend(ctx context.Context, date time.T query := riExchangeDailySpendQuery var total string - err := s.db.QueryRow(ctx, query, date).Scan(&total) + err := s.db.QueryRow(ctx, query, date, nil).Scan(&total) if err != nil { return "", fmt.Errorf("failed to get ri exchange daily spend: %w", err) } @@ -3165,12 +3231,18 @@ func (s *PostgresStore) GetRIExchangeDailySpend(ctx context.Context, date time.T return total, nil } +// riExchangeDailySpendQuery counts completed exchanges accepted inside the +// requested UTC day plus every processing reservation. A processing row is an +// unsettled hold, so it stays counted across UTC midnight until it completes or +// fails. $2 optionally leaves one row out (an approval's own reservation). 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' + WHERE ($2::uuid IS NULL OR id <> $2::uuid) + AND (status = 'processing' + OR (status = 'completed' + 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 diff --git a/internal/config/store_postgres_pgxmock_test.go b/internal/config/store_postgres_pgxmock_test.go index f1c215d17..291b96256 100644 --- a/internal/config/store_postgres_pgxmock_test.go +++ b/internal/config/store_postgres_pgxmock_test.go @@ -1531,7 +1531,7 @@ func TestPGXMock_GetRIExchangeDailySpend_Success(t *testing.T) { ctx := context.Background() rows := pgxmock.NewRows([]string{"total"}).AddRow("250.00") - mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows) + mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg(), nil).WillReturnRows(rows) total, err := store.GetRIExchangeDailySpend(ctx, time.Now()) require.NoError(t, err) @@ -1550,7 +1550,7 @@ func TestPGXMock_GetRIExchangeDailySpend_IncludesProcessingStatus(t *testing.T) // Match any SELECT that goes to the DB; the important assertion is below. rows := pgxmock.NewRows([]string{"total"}).AddRow("0") - mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg()).WillReturnRows(rows) + mock.ExpectQuery("SELECT").WithArgs(pgxmock.AnyArg(), nil).WillReturnRows(rows) _, err := store.GetRIExchangeDailySpend(ctx, time.Now()) require.NoError(t, err) @@ -2240,7 +2240,7 @@ func TestPGXMock_ReserveRIExchange_CommitsCeilingBeforeReturn(t *testing.T) { 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()). + mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg(), 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)) @@ -2260,7 +2260,7 @@ func TestPGXMock_ReserveRIExchange_RollsBackWhenQuoteExceedsHeadroom(t *testing. 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()). + mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg()). WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("101.000000")) mock.ExpectRollback() record := &RIExchangeRecord{PaymentDue: "900", Status: "processing", Mode: "auto"} @@ -2285,7 +2285,7 @@ func TestPGXMock_ReserveRIExchange_DatabaseFailuresDoNotReturnCeiling(t *testing lock.WillReturnError(dbErr) } else { lock.WillReturnResult(pgxmock.NewResult("SELECT", 1)) - read := mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg()) + read := mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg()) if stage == "read" { read.WillReturnError(dbErr) } else { @@ -2312,7 +2312,7 @@ func TestPGXMock_ReserveRIExchange_CommitFailureDoesNotReturnCeiling(t *testing. 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()). + mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg(), pgxmock.AnyArg()). WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("0")) mock.ExpectExec("INSERT INTO ri_exchange_history").WithArgs(anyArgsCfg(21)...). WillReturnResult(pgxmock.NewResult("INSERT", 1)) @@ -3184,3 +3184,57 @@ func TestPGXMock_GetExecutionsByStatuses_ExcludesCleanCompleted(t *testing.T) { assert.Empty(t, execs) assert.NoError(t, mock.ExpectationsWereMet()) } + +func TestPGXMock_ReserveApprovedRIExchange_ReservesCeilingExcludingOwnRow(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + id := "550e8400-e29b-41d4-a716-000000000656" + mock.ExpectBegin() + mock.ExpectQuery("SELECT payment_due::text FROM ri_exchange_history").WithArgs(id). + WillReturnRows(pgxmock.NewRows([]string{"payment_due"}).AddRow("600.000000")) + mock.ExpectExec("SELECT pg_advisory_xact_lock").WithArgs(riExchangeDailySpendLockKey). + WillReturnResult(pgxmock.NewResult("SELECT", 1)) + // The spend read must exclude this row's own id (the legitimate $600 approval). + mock.ExpectQuery("SELECT COALESCE\\(SUM\\(payment_due\\), 0\\)::text").WithArgs(pgxmock.AnyArg(), &id). + WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("0.000000")) + mock.ExpectExec("UPDATE ri_exchange_history SET payment_due").WithArgs(id, "1000.000000"). + WillReturnResult(pgxmock.NewResult("UPDATE", 1)) + mock.ExpectCommit() + + ceiling, err := store.ReserveApprovedRIExchange(context.Background(), id, "1000", "1000") + require.NoError(t, err) + assert.Equal(t, "1000.000000", ceiling) + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestPGXMock_ReserveApprovedRIExchange_RefusesWhenInitialExceedsHeadroom(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + id := "550e8400-e29b-41d4-a716-000000000657" + mock.ExpectBegin() + mock.ExpectQuery("SELECT payment_due::text FROM ri_exchange_history").WithArgs(id). + WillReturnRows(pgxmock.NewRows([]string{"payment_due"}).AddRow("600.000000")) + 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(), &id). + WillReturnRows(pgxmock.NewRows([]string{"spend"}).AddRow("500.000000")) + mock.ExpectRollback() + + _, err := store.ReserveApprovedRIExchange(context.Background(), id, "1000", "1000") + require.ErrorContains(t, err, "daily cap exceeded") + assert.NoError(t, mock.ExpectationsWereMet()) +} + +func TestPGXMock_ReserveApprovedRIExchange_RejectsRowThatIsNotProcessing(t *testing.T) { + mock := newMock(t) + store := storeWith(mock) + const id = "550e8400-e29b-41d4-a716-000000000658" + mock.ExpectBegin() + mock.ExpectQuery("SELECT payment_due::text FROM ri_exchange_history").WithArgs(id). + WillReturnError(pgx.ErrNoRows) + mock.ExpectRollback() + + _, err := store.ReserveApprovedRIExchange(context.Background(), id, "1000", "1000") + require.ErrorContains(t, err, "is not processing") + assert.NoError(t, mock.ExpectationsWereMet()) +} diff --git a/internal/config/store_postgres_ri_exchange_approved_reservation_integration_test.go b/internal/config/store_postgres_ri_exchange_approved_reservation_integration_test.go new file mode 100644 index 000000000..262ba4e9e --- /dev/null +++ b/internal/config/store_postgres_ri_exchange_approved_reservation_integration_test.go @@ -0,0 +1,325 @@ +//go:build integration + +package config + +import ( + "context" + "fmt" + "math/big" + "net/url" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/require" +) + +// Real-PostgreSQL coverage for #656 (approved exchanges reserve the full ceiling +// under the shared lock) and #657 (in-flight holds survive UTC rollover). Uses +// the same dedicated issue41_ database guard as the AUTO reservation test. + +type riExchangeIntegrationEnv struct { + ctx context.Context + poolA, poolB *pgxpool.Pool + storeA, storeB *PostgresStore +} + +func newRIExchangeIntegrationEnv(t *testing.T) *riExchangeIntegrationEnv { + t.Helper() + 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) + t.Cleanup(cancel) + poolA, err := pgxpool.New(ctx, dsn) + require.NoError(t, err) + t.Cleanup(poolA.Close) + poolB, err := pgxpool.New(ctx, dsn) + require.NoError(t, err) + t.Cleanup(poolB.Close) + require.NoError(t, migrations.RunMigrations(ctx, poolA, getTestMigrationsPath(), "", "")) + return &riExchangeIntegrationEnv{ctx: ctx, poolA: poolA, poolB: poolB, + storeA: &PostgresStore{db: poolA}, storeB: &PostgresStore{db: poolB}} +} + +// spend returns the daily spend for date as micros. +func (e *riExchangeIntegrationEnv) spend(t *testing.T, date time.Time) *big.Int { + t.Helper() + text, err := e.storeA.GetRIExchangeDailySpend(e.ctx, date) + require.NoError(t, err) + micros, err := riExchangeMicros(text) + require.NoError(t, err) + return micros +} + +// testAccountID renders a 12-digit account id without a literal that the +// git-secrets AWS account-id pattern would flag. +func testAccountID(n int64) string { return fmt.Sprintf("%012d", n) } + +func usdMicros(dollars int64) *big.Int { + return new(big.Int).Mul(big.NewInt(dollars), riExchangeMicroScale) +} + +// capAbove returns a decimal cap that leaves exactly $1000 of headroom above the +// spend already on the books at date. +func (e *riExchangeIntegrationEnv) capAbove(t *testing.T, date time.Time) string { + t.Helper() + return riExchangeUSD(new(big.Int).Add(e.spend(t, date), usdMicros(1000))) +} + +// processing inserts a processing row holding payment dollars, as a manual +// approval would after the pending->processing transition. +func (e *riExchangeIntegrationEnv) processing(t *testing.T, mode, account string, payment int64, updatedAt time.Time) *RIExchangeRecord { + t.Helper() + record := &RIExchangeRecord{AccountID: account, Region: "us-east-1", SourceRIIDs: []string{"ri-" + account}, + SourceInstanceType: "m5.large", SourceCount: 1, TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: riExchangeUSD(usdMicros(payment)), Status: "processing", Mode: mode} + require.NoError(t, e.storeA.SaveRIExchangeRecord(e.ctx, record)) + e.setUpdatedAt(t, record.ID, updatedAt) + return record +} + +// setUpdatedAt back-dates a row. The ri_exchange_updated_at trigger rewrites +// updated_at on every UPDATE, so it is bypassed for this one statement +// (session_replication_role needs the superuser the dedicated test DB uses). +func (e *riExchangeIntegrationEnv) setUpdatedAt(t *testing.T, id string, updatedAt time.Time) { + t.Helper() + tx, err := e.poolA.Begin(e.ctx) + require.NoError(t, err) + defer tx.Rollback(e.ctx) + _, err = tx.Exec(e.ctx, "SET LOCAL session_replication_role = replica") + require.NoError(t, err) + tag, err := tx.Exec(e.ctx, "UPDATE ri_exchange_history SET updated_at = $2 WHERE id = $1", id, updatedAt) + require.NoError(t, err) + require.EqualValues(t, 1, tag.RowsAffected()) + require.NoError(t, tx.Commit(e.ctx)) + var got time.Time + require.NoError(t, e.poolA.QueryRow(e.ctx, "SELECT updated_at FROM ri_exchange_history WHERE id = $1", id).Scan(&got)) + require.WithinDuration(t, updatedAt, got, time.Millisecond, "back-dating updated_at did not stick") +} + +// release marks a test row failed so it stops holding headroom for later runs. +func (e *riExchangeIntegrationEnv) release(t *testing.T, ids ...string) { + t.Helper() + for _, id := range ids { + require.NoError(t, e.storeA.FailRIExchange(e.ctx, id, "test cleanup")) + } +} + +func TestPostgresStoreDB_ReserveApprovedRIExchange_ConcurrentApprovalsStayWithinCap(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + cap := env.capAbove(t, now) + a := env.processing(t, "manual", "810000000001", 100, now) + b := env.processing(t, "manual", "810000000002", 100, now) + defer env.release(t, a.ID, b.ID) + + type outcome struct { + ceiling string + err error + } + results := make([]outcome, 2) + // Deterministic interleave: hold the global lock from a third transaction so both + // approvals are provably in flight at once, then release it. Without the advisory + // lock in the reservation path neither approval would park, so the wait check fails. + lockTx, err := env.poolA.Begin(env.ctx) + require.NoError(t, err) + defer lockTx.Rollback(env.ctx) + _, err = lockTx.Exec(env.ctx, "SELECT pg_advisory_xact_lock($1)", riExchangeDailySpendLockKey) + require.NoError(t, err) + var wg sync.WaitGroup + for i, run := range []struct { + store *PostgresStore + id string + }{{env.storeA, a.ID}, {env.storeB, b.ID}} { + wg.Add(1) + go func(i int, store *PostgresStore, id string) { + defer wg.Done() + ceiling, reserveErr := store.ReserveApprovedRIExchange(env.ctx, id, cap, "800.000000") + results[i] = outcome{ceiling, reserveErr} + }(i, run.store, run.id) + } + waiters := 0 + for deadline := time.Now().Add(5 * time.Second); time.Now().Before(deadline); time.Sleep(10 * time.Millisecond) { + require.NoError(t, env.poolA.QueryRow(env.ctx, `SELECT COUNT(*) FROM pg_stat_activity + WHERE wait_event_type = 'Lock' AND wait_event = 'advisory' AND query LIKE 'SELECT pg_advisory_xact_lock%'`).Scan(&waiters)) + if waiters >= 2 { + break + } + } + require.Equal(t, 2, waiters, "both approvals must be parked on the global advisory lock at the same time") + for _, id := range []string{a.ID, b.ID} { + var stored string + require.NoError(t, env.poolA.QueryRow(env.ctx, "SELECT payment_due::text FROM ri_exchange_history WHERE id = $1", id).Scan(&stored)) + require.Equal(t, "100.000000", stored, "no approval may reserve before the lock is released") + } + require.NoError(t, lockTx.Commit(env.ctx)) + wg.Wait() + + reserved := new(big.Int) + for _, r := range results { + require.NoError(t, r.err) + micros, parseErr := riExchangeMicros(r.ceiling) + require.NoError(t, parseErr) + reserved.Add(reserved, micros) + } + require.Equal(t, 0, reserved.Cmp(usdMicros(1000)), + "two concurrent approvals must divide, not double, the headroom: %s + %s", results[0].ceiling, results[1].ceiling) + + // The headroom is fully reserved, so a later approval is refused. + c := env.processing(t, "manual", "810000000003", 1, now) + defer env.release(t, c.ID) + _, err = env.storeA.ReserveApprovedRIExchange(env.ctx, c.ID, cap, "800.000000") + require.ErrorContains(t, err, "daily cap exceeded") + + // Each approval settles at its ceiling in the worst case; the day lands exactly on the cap + // (c's refused $1 row is still processing until the handler fails it). + for i, id := range []string{a.ID, b.ID} { + require.NoError(t, env.storeA.CompleteRIExchangeWithPayment(env.ctx, id, "exchange-"+string(rune('a'+i)), results[i].ceiling)) + } + capMicros, err := riExchangeMicros(cap) + require.NoError(t, err) + require.Equal(t, 0, env.spend(t, now).Cmp(new(big.Int).Add(capMicros, usdMicros(1)))) +} + +func TestPostgresStoreDB_ReserveApprovedRIExchange_DoesNotCountOwnInitialQuote(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + cap := env.capAbove(t, now) + row := env.processing(t, "manual", "820000000001", 600, now) + defer env.release(t, row.ID) + + ceiling, err := env.storeA.ReserveApprovedRIExchange(env.ctx, row.ID, cap, "1000.000000") + require.NoError(t, err, "a $600 approval under a $1000 cap must not be rejected for counting itself twice") + require.Equal(t, "1000.000000", ceiling) + + var stored string + require.NoError(t, env.poolB.QueryRow(env.ctx, "SELECT payment_due::text FROM ri_exchange_history WHERE id = $1", row.ID).Scan(&stored)) + require.Equal(t, "1000.000000", stored, "the full ceiling is what the row reserves while it executes") +} + +func TestPostgresStoreDB_ReserveApprovedRIExchange_RejectsRowThatIsNotProcessing(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + row := env.processing(t, "manual", "830000000001", 10, time.Now().UTC()) + env.release(t, row.ID) + + _, err := env.storeA.ReserveApprovedRIExchange(env.ctx, row.ID, env.capAbove(t, time.Now().UTC()), "1000.000000") + require.ErrorContains(t, err, "is not processing") +} + +func TestPostgresStoreDB_ReserveApprovedRIExchange_AutoAndApprovalShareHeadroom(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + cap := env.capAbove(t, now) + approved := env.processing(t, "manual", "840000000001", 100, now) + auto := &RIExchangeRecord{AccountID: testAccountID(840000000002), Region: "eu-west-1", SourceRIIDs: []string{"ri-auto-interleave"}, + SourceInstanceType: "m5.large", SourceCount: 1, TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: "100.000000", Status: "processing", Mode: "auto"} + defer func() { + env.release(t, approved.ID) + if auto.ID != "" { + env.release(t, auto.ID) + } + }() + + var approvedCeiling, autoCeiling string + var approvedErr, autoErr error + start := make(chan struct{}) + var wg sync.WaitGroup + wg.Add(2) + go func() { + defer wg.Done() + <-start + approvedCeiling, approvedErr = env.storeA.ReserveApprovedRIExchange(env.ctx, approved.ID, cap, "800.000000") + }() + go func() { + defer wg.Done() + <-start + autoCeiling, autoErr = env.storeB.ReserveRIExchange(env.ctx, auto, cap, "800.000000") + }() + close(start) + wg.Wait() + require.NoError(t, approvedErr) + require.NoError(t, autoErr) + + approvedMicros, err := riExchangeMicros(approvedCeiling) + require.NoError(t, err) + autoMicros, err := riExchangeMicros(autoCeiling) + require.NoError(t, err) + require.Equal(t, 0, new(big.Int).Add(approvedMicros, autoMicros).Cmp(usdMicros(1000)), + "AUTO and approval reservations must divide, not double, the headroom: %s + %s", approvedCeiling, autoCeiling) +} + +func TestPostgresStoreDB_DailySpend_InFlightReservationSurvivesUTCRollover(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + dayStart := time.Now().UTC().Truncate(24 * time.Hour) + dayB := dayStart.Add(36 * time.Hour) // frozen "next day" read; the query takes the date explicitly + baselineB := env.spend(t, dayB) + + hold := env.processing(t, "auto", "850000000001", 1000, dayStart.Add(time.Minute)) + done := env.processing(t, "auto", "850000000002", 500, dayStart.Add(time.Minute)) + defer env.release(t, hold.ID, done.ID) + require.NoError(t, env.storeA.CompleteRIExchangeWithPayment(env.ctx, done.ID, "exchange-day-a", "500.000000")) + _, err := env.poolA.Exec(env.ctx, "UPDATE ri_exchange_history SET completed_at = $2 WHERE id = $1", done.ID, dayStart.Add(2*time.Minute)) + require.NoError(t, err) + + got := new(big.Int).Sub(env.spend(t, dayB), baselineB) + require.Equal(t, 0, got.Cmp(usdMicros(1000)), + "day B must still count the unsettled day-A hold ($1000) and not the completed day-A spend ($500); delta was %s", riExchangeUSD(got)) +} + +func TestPostgresStoreDB_DailySpend_CompletedRowsFollowAcceptanceDayBoundary(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + dayStart := time.Now().UTC().Truncate(24 * time.Hour) + dayB := dayStart.Add(24 * time.Hour) + baselineA := env.spend(t, dayStart.Add(time.Hour)) + baselineB := env.spend(t, dayB.Add(time.Hour)) + + lastMicroOfA := env.processing(t, "auto", "860000000001", 7, dayStart) + firstInstantOfB := env.processing(t, "auto", "860000000002", 11, dayStart) + defer env.release(t, lastMicroOfA.ID, firstInstantOfB.ID) + require.NoError(t, env.storeA.CompleteRIExchangeWithPayment(env.ctx, lastMicroOfA.ID, "exchange-edge-a", "7.000000")) + require.NoError(t, env.storeA.CompleteRIExchangeWithPayment(env.ctx, firstInstantOfB.ID, "exchange-edge-b", "11.000000")) + _, err := env.poolA.Exec(env.ctx, "UPDATE ri_exchange_history SET completed_at = $2 WHERE id = $1", lastMicroOfA.ID, dayB.Add(-time.Microsecond)) + require.NoError(t, err) + _, err = env.poolA.Exec(env.ctx, "UPDATE ri_exchange_history SET completed_at = $2 WHERE id = $1", firstInstantOfB.ID, dayB) + require.NoError(t, err) + + deltaA := new(big.Int).Sub(env.spend(t, dayStart.Add(time.Hour)), baselineA) + deltaB := new(big.Int).Sub(env.spend(t, dayB.Add(time.Hour)), baselineB) + require.Equal(t, 0, deltaA.Cmp(usdMicros(7)), "00:00:00.999999 belongs to day A only; delta %s", riExchangeUSD(deltaA)) + require.Equal(t, 0, deltaB.Cmp(usdMicros(11)), "exactly 00:00:00 belongs to day B only; delta %s", riExchangeUSD(deltaB)) +} + +func TestPostgresStoreDB_ReserveRIExchange_PriorDayInFlightHoldReducesHeadroom(t *testing.T) { + env := newRIExchangeIntegrationEnv(t) + now := time.Now().UTC() + cap := env.capAbove(t, now) + hold := env.processing(t, "auto", "870000000001", 1000, now.Add(-30*time.Hour)) + defer env.release(t, hold.ID) + + second := &RIExchangeRecord{AccountID: testAccountID(870000000002), Region: "us-east-1", SourceRIIDs: []string{"ri-rollover-second"}, + SourceInstanceType: "m5.large", SourceCount: 1, TargetOfferingID: "offering-test", TargetInstanceType: "m6i.large", TargetCount: 1, + PaymentDue: "1.000000", Status: "processing", Mode: "auto"} + _, err := env.storeA.ReserveRIExchange(env.ctx, second, cap, "1000.000000") + require.ErrorContains(t, err, "daily cap exceeded", "a reservation still executing from yesterday must keep counting today") + require.Empty(t, second.ID) + + // Once the hold settles, only its accepted amount counts, and only on its acceptance day. + require.NoError(t, env.storeA.CompleteRIExchangeWithPayment(env.ctx, hold.ID, "exchange-rollover", "950.000000")) + _, err = env.poolA.Exec(env.ctx, "UPDATE ri_exchange_history SET completed_at = $2 WHERE id = $1", hold.ID, now.Add(-30*time.Hour)) + require.NoError(t, err) + ceiling, err := env.storeA.ReserveRIExchange(env.ctx, second, cap, "1000.000000") + require.NoError(t, err) + require.Equal(t, "1000.000000", ceiling) + env.release(t, second.ID) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index a7caa7a01..afc92cc46 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -666,6 +666,12 @@ func (m *MockConfigStore) ReserveRIExchange(ctx context.Context, record *config. return args.String(0), args.Error(1) } +func (m *MockConfigStore) ReserveApprovedRIExchange(ctx context.Context, id, dailyCapUSD, perExchangeCapUSD string) (string, error) { + m.record("ReserveApprovedRIExchange", ctx, id, dailyCapUSD, perExchangeCapUSD) + args := m.Called(ctx, id, 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/handler_ri_exchange_test.go b/internal/server/handler_ri_exchange_test.go index 18f4c41ef..142985d02 100644 --- a/internal/server/handler_ri_exchange_test.go +++ b/internal/server/handler_ri_exchange_test.go @@ -897,6 +897,10 @@ func (m *mockConfigStoreForExchange) ReserveRIExchange(ctx context.Context, reco return "", errors.New("RI exchange reservation not configured in test store") } +func (m *mockConfigStoreForExchange) ReserveApprovedRIExchange(ctx context.Context, id, dailyCapUSD, perExchangeCapUSD string) (string, error) { + return "", errors.New("approved RI exchange reservation not configured in test store") +} + func (m *mockConfigStoreForExchange) CompleteRIExchangeWithPayment(ctx context.Context, id, exchangeID, acceptedPaymentDue string) error { if m.completeWithPaymentFunc != nil { return m.completeWithPaymentFunc(ctx, id, exchangeID, acceptedPaymentDue) diff --git a/internal/server/test_helpers_test.go b/internal/server/test_helpers_test.go index 5944314b5..7308396ac 100644 --- a/internal/server/test_helpers_test.go +++ b/internal/server/test_helpers_test.go @@ -184,6 +184,10 @@ func (m *mockConfigStoreForHealth) SaveRIExchangeRecord(ctx context.Context, rec 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) ReserveApprovedRIExchange(ctx context.Context, id, dailyCapUSD, perExchangeCapUSD string) (string, error) { + return "", errors.New("approved RI exchange reservation not configured in test store") +} func (m *mockConfigStoreForHealth) GetRIExchangeRecord(ctx context.Context, id string) (*config.RIExchangeRecord, error) { return nil, nil }