diff --git a/internal/api/handler_purchases_revoke.go b/internal/api/handler_purchases_revoke.go index cf307d8d..a3fc2d27 100644 --- a/internal/api/handler_purchases_revoke.go +++ b/internal/api/handler_purchases_revoke.go @@ -950,9 +950,14 @@ func (h *Handler) handleAzureReturnError(ctx context.Context, record *config.Pur // backoff retries. If every attempt fails, Azure has already refunded but the // DB write could not land, so it returns a 207 RECONCILE_PENDING result (no // retry) for the finalize_revocations sweep to reconcile, rather than a 500. +// +// Each attempt first records that Azure accepted the Return +// (MarkRevocationReturnPosted, until one lands): the finalize_revocations +// sweep only finalizes rows with that record (issue #1523). func (h *Handler) persistAzureRevocation(ctx context.Context, record *config.PurchaseHistoryRecord, calcRefundAmount *float64, calcRefundCurrency string) (any, error) { now := time.Now().UTC() - markErr := h.config.MarkPurchaseRevoked(ctx, record.PurchaseID, now, "direct-api", "", calcRefundAmount, calcRefundCurrency) + returnRecorded := false + markErr := h.recordAzureReturnAndMark(ctx, record.PurchaseID, now, calcRefundAmount, calcRefundCurrency, &returnRecorded) for attempt, backoff := range revokeMarkRetryBackoffs { if markErr == nil { break @@ -960,15 +965,16 @@ func (h *Handler) persistAzureRevocation(ctx context.Context, record *config.Pur logging.Warnf("revoke azure: MarkPurchaseRevoked attempt %d failed for %s: %v (retrying in %s)", attempt+1, record.PurchaseID, markErr, backoff) time.Sleep(backoff) - markErr = h.config.MarkPurchaseRevoked(ctx, record.PurchaseID, now, "direct-api", "", calcRefundAmount, calcRefundCurrency) + markErr = h.recordAzureReturnAndMark(ctx, record.PurchaseID, now, calcRefundAmount, calcRefundCurrency, &returnRecorded) } if markErr != nil { // All retries failed. Azure has already refunded but we cannot persist // the revocation state. Return 207 Multi-Status so the frontend knows // the refund issued but not to retry — the finalize_revocations sweep - // will reconcile the DB write on its next tick. - logging.Errorf("revoke azure: MarkPurchaseRevoked failed for %s after %d attempts (Azure already returned): %v", - record.PurchaseID, len(revokeMarkRetryBackoffs)+1, markErr) + // will reconcile the DB write on its next tick, provided the Return + // was recorded. + logging.Errorf("revoke azure: MarkPurchaseRevoked failed for %s after %d attempts (Azure already returned, return recorded=%t): %v", + record.PurchaseID, len(revokeMarkRetryBackoffs)+1, returnRecorded, markErr) return &revokeReconcilePendingResult{ Code: "RECONCILE_PENDING", AzureReturned: true, @@ -986,6 +992,20 @@ func (h *Handler) persistAzureRevocation(ctx context.Context, record *config.Pur }, nil } +// recordAzureReturnAndMark is one persist attempt after Azure accepted the +// Return: stamp return_posted_at unless an earlier attempt already did (a +// failure is logged; the mark may still land), then MarkPurchaseRevoked. +func (h *Handler) recordAzureReturnAndMark(ctx context.Context, purchaseID string, now time.Time, calcRefundAmount *float64, calcRefundCurrency string, returnRecorded *bool) error { + if !*returnRecorded { + if err := h.config.MarkRevocationReturnPosted(ctx, purchaseID); err != nil { + logging.Warnf("revoke azure: MarkRevocationReturnPosted for %s failed (continuing to the mark): %v", purchaseID, err) + } else { + *returnRecorded = true + } + } + return h.config.MarkPurchaseRevoked(ctx, purchaseID, now, "direct-api", "", calcRefundAmount, calcRefundCurrency) +} + // parseAzureReservationIDs extracts the reservation order ID and reservation ID // from an Azure ARM resource path. The purchase_id is stored as the ARM // resource ID at purchase time. diff --git a/internal/api/handler_purchases_revoke_return_posted_test.go b/internal/api/handler_purchases_revoke_return_posted_test.go new file mode 100644 index 00000000..53af25dc --- /dev/null +++ b/internal/api/handler_purchases_revoke_return_posted_test.go @@ -0,0 +1,111 @@ +package api + +import ( + "context" + "errors" + "sync" + "testing" + "time" + + armreservations "github.com/Azure/azure-sdk-for-go/sdk/resourcemanager/reservations/armreservations" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" +) + +// Issue #1523: the finalize_revocations sweep only finalizes rows whose Return +// was recorded as accepted, so the handler must record it after Azure accepts +// the Return and before (or while retrying) MarkPurchaseRevoked. Stub Azure +// clients and a mock store; nothing here calls Azure. + +// callOrder records the store calls in the order the handler makes them. +type callOrder struct { + mu sync.Mutex + calls []string +} + +func (o *callOrder) add(name string) func(mock.Arguments) { + return func(mock.Arguments) { + o.mu.Lock() + defer o.mu.Unlock() + o.calls = append(o.calls, name) + } +} + +func (o *callOrder) get() []string { + o.mu.Lock() + defer o.mu.Unlock() + return append([]string(nil), o.calls...) +} + +func TestCallAzureReturn_RecordsReturnBeforeMark(t *testing.T) { + t.Parallel() + ctx := context.Background() + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + + r := armReservationRecord() + order := &callOrder{} + mockStore.On("MarkRevocationReturnPosted", ctx, r.PurchaseID).Return(nil).Run(order.add("return")).Once() + mockStore.On("MarkPurchaseRevoked", ctx, r.PurchaseID, mock.AnythingOfType("time.Time"), + "direct-api", "", mock.Anything, mock.Anything).Return(nil).Run(order.add("mark")).Once() + + calcClient := &stubCalcRefundClientWithAmount{amount: 10.0, currency: "USD", sessID: "s-ok"} + returnClient := &stubReturnClient{resp: armreservations.ReturnClientPostResponse{}} + h := &Handler{config: mockStore} + result, err := h.callAzureReturn(ctx, calcClient, returnClient, r, "order-abc", "res-xyz", confirmedQuote(10, "USD")) + require.NoError(t, err) + rr, ok := result.(*revokePurchaseResult) + require.True(t, ok) + assert.Equal(t, "revoked", rr.Status) + assert.Equal(t, []string{"return", "mark"}, order.get(), "the accepted Return must be recorded before the mark") +} + +// A Return Azure refused must not be recorded: the sweep would otherwise +// finalize a purchase that was never refunded. +func TestCallAzureReturn_ReturnErrorRecordsNothing(t *testing.T) { + t.Parallel() + ctx := context.Background() + mockStore := new(MockConfigStore) + + r := armReservationRecord() + calcClient := &stubCalcRefundClientWithAmount{amount: 10.0, currency: "USD", sessID: "s-fail"} + returnClient := &stubReturnClient{err: errors.New("dial tcp: connection refused")} + h := &Handler{config: mockStore} + _, err := h.callAzureReturn(ctx, calcClient, returnClient, r, "order-abc", "res-xyz", confirmedQuote(10, "USD")) + require.Error(t, err) + mockStore.AssertNotCalled(t, "MarkRevocationReturnPosted") + mockStore.AssertNotCalled(t, "MarkPurchaseRevoked") +} + +// When the stamp and the mark both fail, every retry retries the stamp too, +// until it lands; after that only the mark is retried. If the stamp lands and +// the mark never does, the 207 is honest: the sweep will finalize the row. +func TestCallAzureReturn_RetriesReturnRecordUntilItLands(t *testing.T) { + // NOT parallel: mutates the package-global revokeMarkRetryBackoffs. + orig := revokeMarkRetryBackoffs + revokeMarkRetryBackoffs = []time.Duration{0, 0, 0} + t.Cleanup(func() { revokeMarkRetryBackoffs = orig }) + + ctx := context.Background() + mockStore := new(MockConfigStore) + t.Cleanup(func() { mockStore.AssertExpectations(t) }) + + r := armReservationRecord() + dbDown := errors.New("db down") + mockStore.On("MarkRevocationReturnPosted", ctx, r.PurchaseID).Return(dbDown).Twice() + mockStore.On("MarkRevocationReturnPosted", ctx, r.PurchaseID).Return(nil).Once() + mockStore.On("MarkPurchaseRevoked", ctx, r.PurchaseID, mock.AnythingOfType("time.Time"), + "direct-api", "", mock.Anything, mock.Anything).Return(dbDown).Times(4) + + calcClient := &stubCalcRefundClientWithAmount{amount: 10.0, currency: "USD", sessID: "s-retry"} + returnClient := &stubReturnClient{resp: armreservations.ReturnClientPostResponse{}} + h := &Handler{config: mockStore} + result, err := h.callAzureReturn(ctx, calcClient, returnClient, r, "order-abc", "res-xyz", confirmedQuote(10, "USD")) + require.NoError(t, err) + pending, ok := result.(*revokeReconcilePendingResult) + require.True(t, ok) + assert.Equal(t, "RECONCILE_PENDING", pending.Code) + mockStore.AssertNumberOfCalls(t, "MarkRevocationReturnPosted", 3) + mockStore.AssertNumberOfCalls(t, "MarkPurchaseRevoked", 4) +} diff --git a/internal/api/nullable_purchase_cost_integration_test.go b/internal/api/nullable_purchase_cost_integration_test.go index 16f544c2..f7af543b 100644 --- a/internal/api/nullable_purchase_cost_integration_test.go +++ b/internal/api/nullable_purchase_cost_integration_test.go @@ -112,7 +112,9 @@ func TestNullablePurchaseCostPostgres(t *testing.T) { require.Len(t, points, 2) } } - _, err = pool.Exec(ctx, `UPDATE purchase_history SET revocation_in_flight=true WHERE purchase_id='unknown-0'`) + // The finalize queue reads return_posted_at (000148, issue #1523). + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 148)) + _, err = pool.Exec(ctx, `UPDATE purchase_history SET revocation_in_flight=true, return_posted_at=now() WHERE purchase_id='unknown-0'`) require.NoError(t, err) inFlight, err := store.GetPurchaseHistoryInFlight(ctx) require.NoError(t, err) diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 65099876..b1ddea0f 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -439,12 +439,28 @@ type StoreInterface interface { // a flag another request set (issue #1518). ClearRevocationInFlight(ctx context.Context, purchaseID string) error - // GetPurchaseHistoryInFlight returns all purchase_history rows with - // revocation_in_flight=true and revoked_at IS NULL. These are rows where - // the Azure Return call succeeded but MarkPurchaseRevoked failed; the - // finalize_revocations scheduled sweep calls this to retry the DB write. + // GetPurchaseHistoryInFlight returns the purchase_history rows whose + // Return was recorded by MarkRevocationReturnPosted but whose revoked_at + // is still NULL: the Azure Return call succeeded but MarkPurchaseRevoked + // failed. The finalize_revocations scheduled sweep calls this to retry the + // DB write. A flagged row with no recorded Return is not returned + // (issue #1523): no refund is known to have gone out for it. GetPurchaseHistoryInFlight(ctx context.Context) ([]*PurchaseHistoryRecord, error) + // MarkRevocationReturnPosted records that the provider accepted the + // Return for purchaseID (return_posted_at, migration 000148). Called by + // the revoke handler after the Azure Return succeeds and before + // MarkPurchaseRevoked. The first stamp stands; a revoked row is a no-op. + // Returns a not-found error when no row matches (issue #1523). + MarkRevocationReturnPosted(ctx context.Context, purchaseID string) error + + // ReleaseStaleRevocationsInFlight clears revocation_in_flight on unrevoked + // rows flagged for longer than after with no Return recorded, so the + // purchase can be revoked again, and returns their purchase IDs. A flagged + // row with no flip timestamp is stamped now and aged from then. Called by + // the finalize_revocations sweep (issue #1523). + ReleaseStaleRevocationsInFlight(ctx context.Context, after time.Duration) ([]string, error) + // UpdatePurchaseHistoryListing stamps the AWS marketplace listing_id and // listing_state onto a purchase_history row. Called after // CreateReservedInstancesListing succeeds (listing_state="active") and diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index fb28430c..c4d9767b 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -3060,9 +3060,12 @@ func (s *PostgresStore) MarkPurchaseRevoked(ctx context.Context, purchaseID stri // row, so exactly one flip of an unrevoked, unflagged row succeeds. A missing // row keeps the not-found error. func (s *PostgresStore) FlipPurchaseRevocationInFlight(ctx context.Context, purchaseID string) error { + // revocation_in_flight_at (migration 000148, issue #1523) is the database + // clock at the flip; the finalize_revocations sweep uses it to release a + // flag whose Return was never recorded. tag, err := s.db.Exec(ctx, ` UPDATE purchase_history - SET revocation_in_flight = true + SET revocation_in_flight = true, revocation_in_flight_at = now() WHERE purchase_id = $1 AND revoked_at IS NULL AND revocation_in_flight = false @@ -3117,19 +3120,27 @@ func (s *PostgresStore) ClearRevocationInFlight(ctx context.Context, purchaseID return nil } -// GetPurchaseHistoryInFlight returns all purchase_history rows with -// revocation_in_flight=true and revoked_at IS NULL. Used by the +// GetPurchaseHistoryInFlight returns the purchase_history rows whose Azure +// Return was recorded as accepted (return_posted_at set by +// MarkRevocationReturnPosted) but whose revoked_at is still NULL. Used by the // finalize_revocations scheduled sweep to retry MarkPurchaseRevoked for rows // where the Azure Return succeeded but the subsequent DB write failed // (issue #290 Finding #6). +// +// revocation_in_flight alone is not enough (issue #1523): a row is also left +// flagged when a request dies between the flip and the Return, and marking +// that row revoked would record a refund that never happened. The flag is not +// required either: once the Return is recorded the purchase is refunded, even +// if a concurrent request's failed Return cleared the flag. func (s *PostgresStore) GetPurchaseHistoryInFlight(ctx context.Context) ([]*PurchaseHistoryRecord, error) { rows, err := s.db.Query(ctx, ` SELECT account_id, purchase_id, timestamp, provider, service, region, resource_type, count, term, payment, upfront_cost, monthly_cost, estimated_savings, plan_id, plan_name, ramp_step, cloud_account_id, - revocation_window_closes_at, revoked_at, revoked_via, support_case_id + revocation_window_closes_at, revoked_at, revoked_via, support_case_id, + revocation_in_flight FROM purchase_history - WHERE revocation_in_flight = true + WHERE return_posted_at IS NOT NULL AND revoked_at IS NULL `) if err != nil { @@ -3167,6 +3178,7 @@ func (s *PostgresStore) GetPurchaseHistoryInFlight(ctx context.Context) ([]*Purc &revokedAt, &revokedVia, &supportCaseID, + &r.RevocationInFlight, ); err != nil { return nil, fmt.Errorf("GetPurchaseHistoryInFlight scan: %w", err) } @@ -3189,7 +3201,6 @@ func (s *PostgresStore) GetPurchaseHistoryInFlight(ctx context.Context) ([]*Purc if supportCaseID.Valid { r.SupportCaseID = supportCaseID.String } - r.RevocationInFlight = true result = append(result, &r) } return result, rows.Err() diff --git a/internal/config/store_postgres_revocation_return.go b/internal/config/store_postgres_revocation_return.go new file mode 100644 index 00000000..e6cc62f9 --- /dev/null +++ b/internal/config/store_postgres_revocation_return.go @@ -0,0 +1,96 @@ +package config + +import ( + "context" + "fmt" + "time" +) + +// MarkRevocationReturnPosted records that the provider accepted the Return for +// purchaseID (issue #1523, migration 000148). The revoke handler calls it right +// after Azure accepts the Return and before MarkPurchaseRevoked, so the +// finalize_revocations sweep can tell "refunded, audit write failed" (finalize) +// from "flagged, but no Return ever went out" (never finalize). +// +// The first stamp stands (COALESCE), so a retry does not move it. A revoked +// row is left alone and returns nil: there is nothing left to finalize. A +// missing row returns a not-found error. +func (s *PostgresStore) MarkRevocationReturnPosted(ctx context.Context, purchaseID string) error { + tag, err := s.db.Exec(ctx, ` + UPDATE purchase_history + SET return_posted_at = COALESCE(return_posted_at, now()) + WHERE purchase_id = $1 + AND revoked_at IS NULL + `, purchaseID) + if err != nil { + return fmt.Errorf("MarkRevocationReturnPosted: %w", err) + } + if tag.RowsAffected() > 0 { + return nil + } + var exists bool + if err := s.db.QueryRow(ctx, `SELECT EXISTS(SELECT 1 FROM purchase_history WHERE purchase_id = $1)`, purchaseID).Scan(&exists); err != nil { + return fmt.Errorf("MarkRevocationReturnPosted existence check: %w", err) + } + if !exists { + return fmt.Errorf("MarkRevocationReturnPosted: purchase_id %q not found", purchaseID) + } + return nil +} + +// ReleaseStaleRevocationsInFlight clears revocation_in_flight on unrevoked +// rows that have been flagged for longer than after with no Return recorded +// (issue #1523), and returns their purchase IDs. Such a row is one whose +// request died between the flip and the Return, or whose Return was never +// recorded as accepted. Clearing the flag lets the user revoke it again; if +// Azure did take a Return that went unrecorded, Azure refuses the second one. +// +// A flagged row with no revocation_in_flight_at (flagged before migration 000148, +// or by a binary that predates it) is stamped now and released once after has +// passed from then, so its age is never overstated. +// +// Both statements use the database clock. Each UPDATE re-checks its predicate +// on the locked row under READ COMMITTED, so a concurrent +// MarkRevocationReturnPosted or a fresh flip that commits first makes the row +// no longer match; one that commits after the release still records its +// Return, which GetPurchaseHistoryInFlight finds with or without the flag. +func (s *PostgresStore) ReleaseStaleRevocationsInFlight(ctx context.Context, after time.Duration) ([]string, error) { + if after <= 0 { + return nil, fmt.Errorf("ReleaseStaleRevocationsInFlight: timeout must be positive, got %s", after) + } + if _, err := s.db.Exec(ctx, ` + UPDATE purchase_history + SET revocation_in_flight_at = now() + WHERE revocation_in_flight + AND revoked_at IS NULL + AND return_posted_at IS NULL + AND revocation_in_flight_at IS NULL + `); err != nil { + return nil, fmt.Errorf("ReleaseStaleRevocationsInFlight stamp: %w", err) + } + rows, err := s.db.Query(ctx, ` + UPDATE purchase_history + SET revocation_in_flight = false, revocation_in_flight_at = NULL + WHERE revocation_in_flight + AND revoked_at IS NULL + AND return_posted_at IS NULL + AND revocation_in_flight_at < now() - make_interval(secs => $1) + RETURNING purchase_id + `, after.Seconds()) + if err != nil { + return nil, fmt.Errorf("ReleaseStaleRevocationsInFlight: %w", err) + } + defer rows.Close() + var released []string + for rows.Next() { + var id string + if err := rows.Scan(&id); err != nil { + return nil, fmt.Errorf("ReleaseStaleRevocationsInFlight scan: %w", err) + } + released = append(released, id) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("ReleaseStaleRevocationsInFlight: %w", err) + } + return released, nil +} diff --git a/internal/config/store_postgres_revocation_return_integration_test.go b/internal/config/store_postgres_revocation_return_integration_test.go new file mode 100644 index 00000000..4041a803 --- /dev/null +++ b/internal/config/store_postgres_revocation_return_integration_test.go @@ -0,0 +1,270 @@ +//go:build integration + +package config + +// Real-DB tests for issue #1523: the finalize_revocations sweep's queue is +// rows with a recorded Azure Return, and a flag with no recorded Return is +// released after a timeout instead of being finalized. + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const testStaleAfter = 2 * time.Hour + +type revocationState struct { + inFlight bool + inFlightAt *time.Time + returnPostedAt *time.Time + revokedAt *time.Time +} + +func readRevocationState(ctx context.Context, t *testing.T, store *PostgresStore, purchaseID string) revocationState { + t.Helper() + var st revocationState + require.NoError(t, store.db.QueryRow(ctx, ` + SELECT revocation_in_flight, revocation_in_flight_at, return_posted_at, revoked_at + FROM purchase_history WHERE purchase_id = $1`, purchaseID). + Scan(&st.inFlight, &st.inFlightAt, &st.returnPostedAt, &st.revokedAt)) + return st +} + +// backdateFlip makes a flagged row look as if it was flipped age ago. +func backdateFlip(ctx context.Context, t *testing.T, store *PostgresStore, purchaseID string, age time.Duration) { + t.Helper() + _, err := store.db.Exec(ctx, ` + UPDATE purchase_history + SET revocation_in_flight_at = now() - make_interval(secs => $2) + WHERE purchase_id = $1`, purchaseID, age.Seconds()) + require.NoError(t, err) +} + +func finalizeQueueIDs(ctx context.Context, t *testing.T, store *PostgresStore) map[string]bool { + t.Helper() + rows, err := store.GetPurchaseHistoryInFlight(ctx) + require.NoError(t, err) + ids := map[string]bool{} + for _, r := range rows { + ids[r.PurchaseID] = true + } + return ids +} + +func TestGetPurchaseHistoryInFlight_RequiresRecordedReturn(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + const flaggedOnly, returned, returnedFlagCleared, revoked = "rr-1523-flagged-only", "rr-1523-returned", "rr-1523-returned-cleared", "rr-1523-revoked" + for _, id := range []string{flaggedOnly, returned, returnedFlagCleared, revoked} { + seedRevocationRow(ctx, t, store, id) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, id)) + } + require.NoError(t, store.MarkRevocationReturnPosted(ctx, returned)) + // A concurrent request's failed Return cleared the flag after this one's + // Return was accepted: the refund still happened and must be finalized. + require.NoError(t, store.MarkRevocationReturnPosted(ctx, returnedFlagCleared)) + require.NoError(t, store.ClearRevocationInFlight(ctx, returnedFlagCleared)) + require.NoError(t, store.MarkRevocationReturnPosted(ctx, revoked)) + require.NoError(t, store.MarkPurchaseRevoked(ctx, revoked, time.Now().UTC(), "direct-api", "", nil, "")) + + rows, err := store.GetPurchaseHistoryInFlight(ctx) + require.NoError(t, err) + got := map[string]bool{} + for _, r := range rows { + got[r.PurchaseID] = r.RevocationInFlight + } + assert.NotContains(t, got, flaggedOnly, "a flag with no recorded Return must never reach the finalize queue") + assert.NotContains(t, got, revoked) + require.Contains(t, got, returned) + assert.True(t, got[returned], "the flag is read from the row") + require.Contains(t, got, returnedFlagCleared) + assert.False(t, got[returnedFlagCleared], "the flag is read from the row, not assumed") +} + +func TestMarkRevocationReturnPosted(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + t.Run("a missing row is not found", func(t *testing.T) { + err := store.MarkRevocationReturnPosted(ctx, "rr-1523-missing") + require.ErrorContains(t, err, "not found") + }) + + t.Run("the first stamp stands", func(t *testing.T) { + const id = "rr-1523-stamp-twice" + seedRevocationRow(ctx, t, store, id) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, id)) + require.NoError(t, store.MarkRevocationReturnPosted(ctx, id)) + first := readRevocationState(ctx, t, store, id).returnPostedAt + require.NotNil(t, first) + time.Sleep(5 * time.Millisecond) + require.NoError(t, store.MarkRevocationReturnPosted(ctx, id)) + again := readRevocationState(ctx, t, store, id).returnPostedAt + require.NotNil(t, again) + assert.True(t, first.Equal(*again)) + }) + + t.Run("a revoked row is left alone", func(t *testing.T) { + const id = "rr-1523-stamp-revoked" + seedRevocationRow(ctx, t, store, id) + require.NoError(t, store.MarkPurchaseRevoked(ctx, id, time.Now().UTC(), "support-case", "case-1", nil, "")) + require.NoError(t, store.MarkRevocationReturnPosted(ctx, id)) + assert.Nil(t, readRevocationState(ctx, t, store, id).returnPostedAt) + }) +} + +func TestReleaseStaleRevocationsInFlight(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + t.Run("a non-positive timeout is refused", func(t *testing.T) { + _, err := store.ReleaseStaleRevocationsInFlight(ctx, 0) + require.Error(t, err) + }) + + t.Run("only stale flags with no recorded Return are released", func(t *testing.T) { + const fresh, almost, stale, staleReturned, staleRevoked = "rr-1523-fresh", "rr-1523-almost", "rr-1523-stale", "rr-1523-stale-returned", "rr-1523-stale-revoked" + for _, id := range []string{fresh, almost, stale, staleReturned, staleRevoked} { + seedRevocationRow(ctx, t, store, id) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, id)) + } + require.NotNil(t, readRevocationState(ctx, t, store, fresh).inFlightAt, "the flip stamps revocation_in_flight_at") + require.NoError(t, store.MarkRevocationReturnPosted(ctx, staleReturned)) + _, err := store.db.Exec(ctx, `UPDATE purchase_history SET revoked_at = now() WHERE purchase_id = $1`, staleRevoked) + require.NoError(t, err) + for _, id := range []string{stale, staleReturned, staleRevoked} { + backdateFlip(ctx, t, store, id, testStaleAfter+time.Minute) + } + backdateFlip(ctx, t, store, almost, testStaleAfter-time.Minute) + + released, err := store.ReleaseStaleRevocationsInFlight(ctx, testStaleAfter) + require.NoError(t, err) + assert.ElementsMatch(t, []string{stale}, released) + + st := readRevocationState(ctx, t, store, stale) + assert.False(t, st.inFlight, "the stale flag is cleared") + assert.Nil(t, st.inFlightAt) + assert.Nil(t, st.revokedAt, "a released row is never marked revoked") + assert.Nil(t, st.returnPostedAt) + assert.True(t, readRevocationState(ctx, t, store, fresh).inFlight, "a flag younger than the timeout stays") + assert.True(t, readRevocationState(ctx, t, store, almost).inFlight, "a flag just under the timeout stays") + assert.True(t, readRevocationState(ctx, t, store, staleReturned).inFlight, "a recorded Return is finalized, not released") + assert.True(t, readRevocationState(ctx, t, store, staleRevoked).inFlight, "revoked rows are not the release's business") + + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, stale), "a released row can be revoked again") + }) + + // A row flagged before migration 000148 (or by a binary without it) has no + // flip timestamp. Its age is unknown, so it is stamped on first sight and + // released only once the timeout has passed from then. + t.Run("a legacy flag is aged from first sight", func(t *testing.T) { + const id = "rr-1523-legacy" + seedRevocationRow(ctx, t, store, id) + _, err := store.db.Exec(ctx, `UPDATE purchase_history SET revocation_in_flight = true WHERE purchase_id = $1`, id) + require.NoError(t, err) + require.Nil(t, readRevocationState(ctx, t, store, id).inFlightAt) + + released, err := store.ReleaseStaleRevocationsInFlight(ctx, testStaleAfter) + require.NoError(t, err) + assert.NotContains(t, released, id) + st := readRevocationState(ctx, t, store, id) + assert.True(t, st.inFlight) + require.NotNil(t, st.inFlightAt, "first sight stamps the row") + assert.NotContains(t, finalizeQueueIDs(ctx, t, store), id, "a legacy flag is never finalized") + + backdateFlip(ctx, t, store, id, testStaleAfter+time.Minute) + released, err = store.ReleaseStaleRevocationsInFlight(ctx, testStaleAfter) + require.NoError(t, err) + assert.Contains(t, released, id) + assert.False(t, readRevocationState(ctx, t, store, id).inFlight) + }) +} + +// The handler's MarkRevocationReturnPosted can race the sweep's release of the +// same stale row. Whichever commits first, a recorded Return must end in the +// finalize queue: the release re-checks return_posted_at IS NULL on the locked +// row, and the queue does not require the flag. +func TestReleaseStaleRevocationsInFlight_RacesReturnRecord(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + releasedRounds := 0 + for round := range 30 { + id := fmt.Sprintf("rr-1523-race-return-%d", round) + seedRevocationRow(ctx, t, store, id) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, id)) + backdateFlip(ctx, t, store, id, testStaleAfter+time.Minute) + + var wg sync.WaitGroup + var stampErr, releaseErr error + var released []string + start := make(chan struct{}) + wg.Add(2) + go func() { defer wg.Done(); <-start; stampErr = store.MarkRevocationReturnPosted(ctx, id) }() + go func() { + defer wg.Done() + <-start + released, releaseErr = store.ReleaseStaleRevocationsInFlight(ctx, testStaleAfter) + }() + close(start) + wg.Wait() + require.NoError(t, stampErr) + require.NoError(t, releaseErr) + + st := readRevocationState(ctx, t, store, id) + require.NotNil(t, st.returnPostedAt, "round %d", round) + assert.Nil(t, st.revokedAt, "round %d: the release never revokes", round) + if len(released) > 0 { + releasedRounds++ + assert.False(t, st.inFlight, "round %d", round) + } else { + assert.True(t, st.inFlight, "round %d", round) + } + require.True(t, finalizeQueueIDs(ctx, t, store)[id], "round %d: a recorded Return must reach the finalize queue (released=%v)", round, released) + require.NoError(t, store.MarkPurchaseRevoked(ctx, id, time.Now().UTC(), "direct-api", "", nil, "")) + } + t.Logf("release won %d of 30 rounds", releasedRounds) +} + +// A new flip can race the release of a stale flag on the same row. A flip that +// succeeds must leave the row flagged with a fresh stamp whichever commits +// first: the release re-checks the stamp's age on the locked row. +func TestReleaseStaleRevocationsInFlight_RacesFlip(t *testing.T) { + ctx := context.Background() + store := setupRampStepStore(ctx, t) + + for round := range 30 { + id := fmt.Sprintf("rr-1523-race-flip-%d", round) + seedRevocationRow(ctx, t, store, id) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, id)) + backdateFlip(ctx, t, store, id, testStaleAfter+time.Minute) + + var wg sync.WaitGroup + var flipErr, releaseErr error + start := make(chan struct{}) + wg.Add(2) + go func() { defer wg.Done(); <-start; flipErr = store.FlipPurchaseRevocationInFlight(ctx, id) }() + go func() { + defer wg.Done() + <-start + _, releaseErr = store.ReleaseStaleRevocationsInFlight(ctx, testStaleAfter) + }() + close(start) + wg.Wait() + require.NoError(t, releaseErr) + if flipErr != nil { + continue // a flip that refused the row (an in-flight claim) posts no Return + } + st := readRevocationState(ctx, t, store, id) + assert.True(t, st.inFlight, "round %d: a successful flip must not be released", round) + require.NotNil(t, st.inFlightAt, "round %d", round) + assert.WithinDuration(t, time.Now(), *st.inFlightAt, time.Minute, "round %d: the flip's stamp is fresh", round) + } +} diff --git a/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted.down.sql b/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted.down.sql new file mode 100644 index 00000000..399110d7 --- /dev/null +++ b/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted.down.sql @@ -0,0 +1,8 @@ +-- Reverse 000148 (issue #1523). Lossy: which unrevoked rows had an accepted +-- Return recorded is dropped with the column. revocation_in_flight is left as +-- it is, so the pre-000148 sweep will mark every flagged, unrevoked row +-- revoked again. +DROP INDEX IF EXISTS idx_purchase_history_return_posted_unrevoked; +ALTER TABLE purchase_history + DROP COLUMN IF EXISTS revocation_in_flight_at, + DROP COLUMN IF EXISTS return_posted_at; diff --git a/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted.up.sql b/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted.up.sql new file mode 100644 index 00000000..8c05a6f2 --- /dev/null +++ b/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted.up.sql @@ -0,0 +1,30 @@ +-- Record whether an Azure Return was actually accepted (issue #1523). +-- +-- The finalize_revocations sweep used to mark every flagged, unrevoked +-- purchase_history row revoked. A row is also left flagged when a request +-- crashes after FlipPurchaseRevocationInFlight but before Azure accepts the +-- Return (or when the flip's commit result is lost), so the sweep could mark a +-- purchase revoked that was never refunded and is still billed. +-- +-- return_posted_at stamped by the revoke handler right after Azure +-- accepts the Return, before MarkPurchaseRevoked. +-- The sweep finalizes only rows that have it. +-- revocation_in_flight_at stamped by FlipPurchaseRevocationInFlight. A row +-- still flagged with no return_posted_at long after +-- this has had its flag cleared by the sweep, so +-- the purchase can be revoked again. +-- +-- Rows already flagged when this lands are not backfilled: no Return was +-- recorded for them, so they are treated as "Return not recorded". The sweep +-- never finalizes them. It stamps revocation_in_flight_at the first time it +-- sees such a row and clears the flag once the timeout has passed from then. +-- Nothing here changes revocation_in_flight or revoked_at. + +ALTER TABLE purchase_history + ADD COLUMN IF NOT EXISTS return_posted_at TIMESTAMPTZ, + ADD COLUMN IF NOT EXISTS revocation_in_flight_at TIMESTAMPTZ; + +-- The sweep's finalize queue: Return accepted, revocation not yet recorded. +CREATE INDEX IF NOT EXISTS idx_purchase_history_return_posted_unrevoked + ON purchase_history (purchase_id) + WHERE return_posted_at IS NOT NULL AND revoked_at IS NULL; diff --git a/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted_test.go b/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted_test.go new file mode 100644 index 00000000..f25d9c24 --- /dev/null +++ b/internal/database/postgres/migrations/000148_purchase_history_revocation_return_posted_test.go @@ -0,0 +1,80 @@ +//go:build integration +// +build integration + +package migrations_test + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/migrations" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// TestMigration_000148_RevocationReturnPosted covers the schema half of issue +// #1523. Rows flagged in flight before 000148 have no record of an accepted +// Return, so the up must not invent one (a fabricated return_posted_at would +// let the sweep mark a possibly unrefunded purchase revoked) and must not +// touch the flag or revoked_at. The down drops both columns and leaves the +// flag alone. +func TestMigration_000148_RevocationReturnPosted(t *testing.T) { + ctx := context.Background() + path := getMigrationsPath() + pool := testhelpers.RequireMigratedDatabase(ctx, t, path).DB.Pool() + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, previousMigrationVersion(t, 148))) + + revokedAt := time.Now().UTC() + rows := []struct { + id string + inFlight bool + revokedAt *time.Time + }{ + {"ri-148-flagged", true, nil}, + {"ri-148-revoked", false, &revokedAt}, + {"ri-148-active", false, nil}, + } + for _, r := range rows { + _, err := pool.Exec(ctx, ` + INSERT INTO purchase_history ( + account_id, purchase_id, timestamp, provider, service, region, + resource_type, term, payment, revocation_in_flight, revoked_at + ) VALUES ('acct', $1, NOW(), 'azure', 'compute', 'eastus', 'Standard_D2s_v3', 12, 'All Upfront', $2, $3)`, + r.id, r.inFlight, r.revokedAt) + require.NoError(t, err) + } + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 148)) + + for _, r := range rows { + var inFlight bool + var revoked, returnPosted, inFlightAt *time.Time + require.NoError(t, pool.QueryRow(ctx, ` + SELECT revocation_in_flight, revoked_at, return_posted_at, revocation_in_flight_at + FROM purchase_history WHERE purchase_id = $1`, r.id).Scan(&inFlight, &revoked, &returnPosted, &inFlightAt)) + assert.Equal(t, r.inFlight, inFlight, "%s: the flag is untouched", r.id) + assert.Equal(t, r.revokedAt != nil, revoked != nil, "%s: revoked_at is untouched", r.id) + assert.Nil(t, returnPosted, "%s: no Return is recorded for a pre-000148 row", r.id) + assert.Nil(t, inFlightAt, "%s: the sweep, not the migration, starts a legacy flag's clock", r.id) + } + var indexed int + require.NoError(t, pool.QueryRow(ctx, + `SELECT count(*) FROM pg_indexes WHERE indexname = 'idx_purchase_history_return_posted_unrevoked'`).Scan(&indexed)) + assert.Equal(t, 1, indexed) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, previousMigrationVersion(t, 148))) + var cols int + require.NoError(t, pool.QueryRow(ctx, ` + SELECT count(*) FROM information_schema.columns + WHERE table_name = 'purchase_history' + AND column_name IN ('return_posted_at', 'revocation_in_flight_at')`).Scan(&cols)) + assert.Zero(t, cols, "the down drops both columns") + var flagged int + require.NoError(t, pool.QueryRow(ctx, + `SELECT count(*) FROM purchase_history WHERE revocation_in_flight AND purchase_id LIKE 'ri-148-%'`).Scan(&flagged)) + assert.Equal(t, 1, flagged, "the down leaves the flag alone") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, path, 148), "the up re-applies after a rollback") +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 4a8625e8..ea6706bb 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -914,6 +914,36 @@ func (m *MockConfigStore) GetPurchaseHistoryInFlight(ctx context.Context) ([]*co return v, args.Error(1) } +// MarkRevocationReturnPosted mocks MarkRevocationReturnPosted (issue #1523). +// Uses the isExpected default-or-dispatch pattern: revoke tests that only +// verify MarkPurchaseRevoked do not need an expectation for this stamp. +func (m *MockConfigStore) MarkRevocationReturnPosted(ctx context.Context, purchaseID string) error { + m.record("MarkRevocationReturnPosted", ctx, purchaseID) + if !isExpected(&m.Mock, "MarkRevocationReturnPosted") { + return nil + } + args := m.Called(ctx, purchaseID) + return args.Error(0) +} + +// ReleaseStaleRevocationsInFlight mocks ReleaseStaleRevocationsInFlight +// (issue #1523). Default-or-dispatch: with no expectation it releases nothing. +func (m *MockConfigStore) ReleaseStaleRevocationsInFlight(ctx context.Context, after time.Duration) ([]string, error) { + m.record("ReleaseStaleRevocationsInFlight", ctx, after) + if !isExpected(&m.Mock, "ReleaseStaleRevocationsInFlight") { + return nil, nil + } + args := m.Called(ctx, after) + if args.Get(0) == nil { + return nil, args.Error(1) + } + v, ok := args.Get(0).([]string) + if !ok { + m.reporter.wrongReturnType("ReleaseStaleRevocationsInFlight", "[]string", args.Get(0)) + } + return v, args.Error(1) +} + // UpdatePurchaseHistoryListing mocks stamping listing_id and listing_state (issue #292). func (m *MockConfigStore) UpdatePurchaseHistoryListing(ctx context.Context, purchaseID, listingID, listingState string) error { m.record("UpdatePurchaseHistoryListing", ctx, purchaseID, listingID, listingState) diff --git a/internal/purchase/finalize_revocations.go b/internal/purchase/finalize_revocations.go index e89ee340..b5451705 100644 --- a/internal/purchase/finalize_revocations.go +++ b/internal/purchase/finalize_revocations.go @@ -2,6 +2,7 @@ package purchase import ( "context" + "errors" "fmt" "time" @@ -11,15 +12,28 @@ import ( // FinalizeResult summarizes one sweep of FinalizeInFlightRevocations. // Returned for the scheduled-task handler to log and surface in CloudWatch. type FinalizeResult struct { - // Found is the number of purchase_history rows with revocation_in_flight=true - // and revoked_at IS NULL that the sweep saw. + // Found is the number of unrevoked purchase_history rows with a recorded + // Return (return_posted_at) that the sweep saw. Found int // Finalized is the number of rows successfully stamped by MarkPurchaseRevoked. Finalized int // Errored is the number of rows the MarkPurchaseRevoked call failed for. Errored int + // Released is the number of rows whose revocation_in_flight flag was + // cleared because no Return was recorded within + // staleRevocationInFlightAfter (issue #1523). They are not revoked. + Released int } +// staleRevocationInFlightAfter is how long a row may stay flagged with no +// recorded Return before the sweep clears the flag so the purchase can be +// revoked again (issue #1523). It is twice the longest request a deployment +// can run (Cloud Run's request timeout caps at 3600s; Lambda at 900s), so a +// revoke still between its flip and its Return is never released. Releasing +// one early would not finalize anything wrongly either: a Return recorded +// after the release is still finalized, flag or not. +const staleRevocationInFlightAfter = 2 * time.Hour + // finalizeRevocationBackoffs are the sleep durations between consecutive // MarkPurchaseRevoked retry attempts in FinalizeInFlightRevocations. var finalizeRevocationBackoffs = []time.Duration{ @@ -27,14 +41,22 @@ var finalizeRevocationBackoffs = []time.Duration{ 6 * time.Second, } -// FinalizeInFlightRevocations sweeps purchase_history rows with -// revocation_in_flight=true and revoked_at IS NULL and retries +// FinalizeInFlightRevocations sweeps unrevoked purchase_history rows whose +// Azure Return was recorded as accepted (return_posted_at) and retries // MarkPurchaseRevoked for each. These rows represent cases where the Azure // Return API call succeeded but the subsequent DB write failed — the // finalize_revocations scheduled tick ensures the audit record is eventually // consistent without requiring the user to retry (which would be rejected by // Azure as "already returned"). // +// A row that is only flagged (revocation_in_flight) with no recorded Return is +// never marked revoked (issue #1523): its request may have died before the +// Return was posted, so the purchase may still be active and billed. Once it +// has been flagged for staleRevocationInFlightAfter, the sweep clears the flag +// instead, so the user can revoke it again, and logs it for an operator. A +// failure to release is logged, does not stop the finalize pass, and is +// returned with the result. +// // The sweep uses a fixed timestamp of time.Now() at the START of the sweep so // all rows finalized in one pass share the same revokedAt wall-clock value, // which makes log correlation easier. @@ -49,12 +71,14 @@ var finalizeRevocationBackoffs = []time.Duration{ // the sweep returns them with an error wrapping ctx.Err(), so a Lambda near // its deadline stops instead of sleeping past it (issue #243). func (m *Manager) FinalizeInFlightRevocations(ctx context.Context) (*FinalizeResult, error) { + released, releaseErr := m.releaseStaleRevocations(ctx) + rows, err := m.config.GetPurchaseHistoryInFlight(ctx) if err != nil { - return nil, err + return nil, errors.Join(releaseErr, err) } - result := &FinalizeResult{Found: len(rows)} + result := &FinalizeResult{Found: len(rows), Released: released} now := time.Now().UTC() for _, record := range rows { @@ -63,7 +87,7 @@ func (m *Manager) FinalizeInFlightRevocations(ctx context.Context) (*FinalizeRes result.Errored++ logging.Errorf("finalize_revocations: sweep stopped at purchase_id=%s: %v (found=%d finalized=%d errored=%d)", record.PurchaseID, ctxErr, result.Found, result.Finalized, result.Errored) - return result, fmt.Errorf("finalize_revocations: sweep canceled: %w", ctxErr) + return result, errors.Join(releaseErr, fmt.Errorf("finalize_revocations: sweep canceled: %w", ctxErr)) } if markErr != nil { logging.Errorf("finalize_revocations: MarkPurchaseRevoked for %s failed after %d attempts: %v", @@ -75,7 +99,25 @@ func (m *Manager) FinalizeInFlightRevocations(ctx context.Context) (*FinalizeRes } } - return result, nil + return result, releaseErr +} + +// releaseStaleRevocations clears the flag on rows flagged for longer than +// staleRevocationInFlightAfter with no recorded Return, and logs each one: if +// Azure did take a Return that was never recorded, the row now reads as +// active and an operator has to reconcile it by hand. +func (m *Manager) releaseStaleRevocations(ctx context.Context) (int, error) { + ids, err := m.config.ReleaseStaleRevocationsInFlight(ctx, staleRevocationInFlightAfter) + if err != nil { + logging.Errorf("finalize_revocations: releasing stale in-flight revocations failed: %v", err) + return 0, fmt.Errorf("finalize_revocations: release stale: %w", err) + } + for _, id := range ids { + logging.Errorf("finalize_revocations: purchase_id=%s was flagged in flight for over %s with no recorded Azure Return; "+ + "cleared the flag without marking it revoked so it can be revoked again. If Azure did accept a Return, reconcile the row by hand", + id, staleRevocationInFlightAfter) + } + return len(ids), nil } // markRevokedWithRetry calls MarkPurchaseRevoked, retrying after each diff --git a/internal/purchase/finalize_revocations_integration_test.go b/internal/purchase/finalize_revocations_integration_test.go new file mode 100644 index 00000000..0109255f --- /dev/null +++ b/internal/purchase/finalize_revocations_integration_test.go @@ -0,0 +1,100 @@ +//go:build integration + +package purchase + +// Real PostgreSQL, no cloud calls. Issue #1523: the finalize_revocations sweep +// marked every flagged, unrevoked row revoked, including rows whose request +// died between the flip and the Azure Return, so a purchase that was never +// refunded read back as revoked while it was still active and billed. + +import ( + "context" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func seedAzurePurchase(ctx context.Context, t *testing.T, store *config.PostgresStore, purchaseID string) { + t.Helper() + require.NoError(t, store.SavePurchaseHistory(ctx, &config.PurchaseHistoryRecord{ + AccountID: "acct", PurchaseID: purchaseID, Timestamp: time.Now(), Provider: "azure", Service: "compute", + Region: "eastus", ResourceType: "Standard_D2s_v3", Count: 1, Term: 1, Payment: "All Upfront", + })) +} + +func loadPurchase(ctx context.Context, t *testing.T, store *config.PostgresStore, purchaseID string) *config.PurchaseHistoryRecord { + t.Helper() + got, err := store.GetPurchaseHistoryByPurchaseID(ctx, purchaseID) + require.NoError(t, err) + require.NotNil(t, got) + return got +} + +// Acceptance criterion 1: a row flagged without a recorded Return is not +// marked revoked by the sweep. Both shapes are covered: a request that died +// after the flip (the flip stamped the row), and a row flagged before +// migration 000148 (no stamp at all). +func TestFinalizeInFlightRevocations_FlagWithoutReturnIsNotRevoked(t *testing.T) { + ctx := context.Background() + store, pool := rampStepStore(ctx, t) + m := &Manager{config: store} + + const crashed, legacy = "azure-1523-crashed-before-return", "azure-1523-legacy-flag" + seedAzurePurchase(ctx, t, store, crashed) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, crashed)) + seedAzurePurchase(ctx, t, store, legacy) + _, err := pool.Exec(ctx, `UPDATE purchase_history SET revocation_in_flight = true WHERE purchase_id = $1`, legacy) + require.NoError(t, err) + + res, err := m.FinalizeInFlightRevocations(ctx) + require.NoError(t, err) + assert.Zero(t, res.Finalized, "no Return was recorded, so nothing may be finalized") + + for _, id := range []string{crashed, legacy} { + got := loadPurchase(ctx, t, store, id) + assert.Nil(t, got.RevokedAt, "%s: a purchase with no recorded Return must not read back as revoked", id) + assert.True(t, got.RevocationInFlight, "%s: a fresh flag stays until the timeout", id) + } +} + +// Acceptance criterion 2: a row whose Return was recorded is still finalized, +// as before. And once a flag with no recorded Return has outlived the +// timeout, the sweep clears it (still unrevoked) so the purchase can be +// revoked again. +func TestFinalizeInFlightRevocations_RecordedReturnFinalizedStaleFlagReleased(t *testing.T) { + ctx := context.Background() + store, pool := rampStepStore(ctx, t) + m := &Manager{config: store} + + const returned, stale = "azure-1523-returned", "azure-1523-stale-flag" + seedAzurePurchase(ctx, t, store, returned) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, returned)) + require.NoError(t, store.MarkRevocationReturnPosted(ctx, returned)) + seedAzurePurchase(ctx, t, store, stale) + require.NoError(t, store.FlipPurchaseRevocationInFlight(ctx, stale)) + _, err := pool.Exec(ctx, ` + UPDATE purchase_history + SET revocation_in_flight_at = now() - make_interval(secs => $2) + WHERE purchase_id = $1`, stale, (staleRevocationInFlightAfter + time.Minute).Seconds()) + require.NoError(t, err) + + res, err := m.FinalizeInFlightRevocations(ctx) + require.NoError(t, err) + assert.Equal(t, FinalizeResult{Found: 1, Finalized: 1, Released: 1}, *res) + + got := loadPurchase(ctx, t, store, returned) + require.NotNil(t, got.RevokedAt, "a recorded Return is finalized") + assert.Equal(t, "direct-api", got.RevokedVia) + assert.False(t, got.RevocationInFlight) + + got = loadPurchase(ctx, t, store, stale) + assert.Nil(t, got.RevokedAt, "a released row is not revoked") + assert.False(t, got.RevocationInFlight, "the stale flag is cleared so the user can retry") + + res, err = m.FinalizeInFlightRevocations(ctx) + require.NoError(t, err) + assert.Equal(t, FinalizeResult{}, *res, "a second tick finds nothing left to do") +} diff --git a/internal/purchase/finalize_revocations_test.go b/internal/purchase/finalize_revocations_test.go index aa0d9b97..25d34d25 100644 --- a/internal/purchase/finalize_revocations_test.go +++ b/internal/purchase/finalize_revocations_test.go @@ -101,3 +101,61 @@ func TestFinalizeInFlightRevocations_CanceledDuringBackoffStopsSweep(t *testing. assert.Less(t, elapsed, time.Second, "the sweep must not sleep through the backoff after cancellation") store.AssertNumberOfCalls(t, "MarkPurchaseRevoked", 2) } + +// Issue #1523: before the finalize pass the sweep releases flags that stayed +// set past staleRevocationInFlightAfter with no recorded Return. Released rows +// are counted but never marked revoked. +func TestFinalizeInFlightRevocations_ReleasesStaleFlagsWithoutMarking(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + store.On("ReleaseStaleRevocationsInFlight", ctx, staleRevocationInFlightAfter). + Return([]string{"crashed-before-return", "legacy"}, nil).Once() + store.On("GetPurchaseHistoryInFlight", ctx).Return(inFlightRows("returned"), nil).Once() + expectMark(store, "returned", nil) + m := &Manager{config: store} + + res, err := m.FinalizeInFlightRevocations(ctx) + require.NoError(t, err) + assert.Equal(t, FinalizeResult{Found: 1, Finalized: 1, Released: 2}, *res) + store.AssertExpectations(t) + store.AssertNumberOfCalls(t, "MarkPurchaseRevoked", 1) +} + +// A release failure must not stop the finalize pass (the two queues are +// disjoint), but the tick must still fail so the outage is visible. +func TestFinalizeInFlightRevocations_ReleaseErrorStillFinalizes(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + store.On("ReleaseStaleRevocationsInFlight", ctx, staleRevocationInFlightAfter). + Return(nil, errors.New("release failed")).Once() + store.On("GetPurchaseHistoryInFlight", ctx).Return(inFlightRows("returned"), nil).Once() + expectMark(store, "returned", nil) + m := &Manager{config: store} + + res, err := m.FinalizeInFlightRevocations(ctx) + require.ErrorContains(t, err, "release failed") + require.NotNil(t, res) + assert.Equal(t, FinalizeResult{Found: 1, Finalized: 1}, *res) + store.AssertExpectations(t) +} + +// Both a release failure and a list failure are reported. +func TestFinalizeInFlightRevocations_ReleaseAndListErrorsJoined(t *testing.T) { + ctx := context.Background() + store := new(MockConfigStore) + store.On("ReleaseStaleRevocationsInFlight", ctx, staleRevocationInFlightAfter). + Return(nil, errors.New("release failed")).Once() + store.On("GetPurchaseHistoryInFlight", ctx).Return(nil, errors.New("list failed")).Once() + m := &Manager{config: store} + + res, err := m.FinalizeInFlightRevocations(ctx) + require.ErrorContains(t, err, "release failed") + require.ErrorContains(t, err, "list failed") + assert.Nil(t, res) +} + +// The timeout must outlast any request still between its flip and its Return: +// Cloud Run's request timeout caps at an hour. +func TestStaleRevocationInFlightAfter_OutlastsLongestRequest(t *testing.T) { + assert.Greater(t, staleRevocationInFlightAfter, time.Hour) +} diff --git a/internal/server/handler.go b/internal/server/handler.go index 3d37092b..429dbbf8 100644 --- a/internal/server/handler.go +++ b/internal/server/handler.go @@ -62,8 +62,9 @@ const ( // issue #291 wave-2). Wires the "fire_scheduled_purchases" event action // to purchase.Manager.FireScheduledDelayedPurchases. TaskFireScheduledPurchases ScheduledTaskType = "fire_scheduled_purchases" - // TaskFinalizeRevocations sweeps purchase_history rows with - // revocation_in_flight=true and retries MarkPurchaseRevoked for each. + // TaskFinalizeRevocations sweeps unrevoked purchase_history rows with a + // recorded Azure Return and retries MarkPurchaseRevoked for each (a row + // that is only flagged in flight is never marked revoked, issue #1523). // These rows represent partial-success cases where the Azure Return API call // succeeded but the subsequent DB write failed. The sweep ensures the audit // record is eventually consistent without requiring the user to retry (which @@ -358,11 +359,13 @@ func (app *Application) handleFireScheduledPurchases(ctx context.Context) (*purc return result, nil } -// handleFinalizeRevocations sweeps purchase_history rows with -// revocation_in_flight=true and retries MarkPurchaseRevoked for each. -// Part of the partial-success reconciliation for Azure revocations (issue #290 -// Finding #6): if the Azure Return call succeeded but the DB write failed, -// this sweep ensures the audit row is eventually consistent. +// handleFinalizeRevocations sweeps unrevoked purchase_history rows with a +// recorded Azure Return and retries MarkPurchaseRevoked for each, and clears +// the in-flight flag on rows flagged too long with no recorded Return +// (issue #1523). Part of the partial-success reconciliation for Azure +// revocations (issue #290 Finding #6): if the Azure Return call succeeded but +// the DB write failed, this sweep ensures the audit row is eventually +// consistent. func (app *Application) handleFinalizeRevocations(ctx context.Context) (*purchase.FinalizeResult, error) { log.Println("Finalizing in-flight revocations...") result, err := app.Purchase.FinalizeInFlightRevocations(ctx) @@ -370,8 +373,8 @@ func (app *Application) handleFinalizeRevocations(ctx context.Context) (*purchas log.Printf("Failed to finalize in-flight revocations: %v", err) return nil, err } - log.Printf("Finalize sweep complete: found=%d finalized=%d errored=%d", - result.Found, result.Finalized, result.Errored) + log.Printf("Finalize sweep complete: found=%d finalized=%d errored=%d released=%d", + result.Found, result.Finalized, result.Errored, result.Released) return result, nil } diff --git a/internal/server/handler_marketplace_poll_test.go b/internal/server/handler_marketplace_poll_test.go index 99c09950..f5871da5 100644 --- a/internal/server/handler_marketplace_poll_test.go +++ b/internal/server/handler_marketplace_poll_test.go @@ -47,7 +47,7 @@ func TestFinalizeSweep_PollsMarketplaceListings(t *testing.T) { assert.Equal(t, 1, poller.calls) encoded, err := json.Marshal(out) require.NoError(t, err) - assert.JSONEq(t, `{"Found":1,"Finalized":1,"Errored":0, + assert.JSONEq(t, `{"Found":1,"Finalized":1,"Errored":0,"Released":0, "marketplace_listings":{"found":3,"active":1,"settled":2,"unpriced":0,"skipped":0,"errored":0}}`, string(encoded)) } diff --git a/internal/server/interfaces.go b/internal/server/interfaces.go index b771b956..bae10f86 100644 --- a/internal/server/interfaces.go +++ b/internal/server/interfaces.go @@ -55,8 +55,9 @@ type PurchaseManagerInterface interface { // whose scheduled_execution_at is in the past (Gmail-style pre-fire delay, // issue #291 wave-2). Called on the "fire_scheduled_purchases" scheduler tick. FireScheduledDelayedPurchases(ctx context.Context) (*purchase.FireResult, error) - // FinalizeInFlightRevocations sweeps purchase_history rows with - // revocation_in_flight=true and retries MarkPurchaseRevoked for each. Handles + // FinalizeInFlightRevocations sweeps unrevoked purchase_history rows with a + // recorded Azure Return and retries MarkPurchaseRevoked for each, and + // releases flags that stayed set with no recorded Return (issue #1523). Handles // the partial-success case where Azure Return succeeded but the DB write // failed (issue #290 Finding #6). Called on the "finalize_revocations" tick. FinalizeInFlightRevocations(ctx context.Context) (*purchase.FinalizeResult, error)