From d1dd5af54cb5b5d57ef551b4a15c9c56bf8161e2 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Mon, 26 Oct 2026 19:30:34 +0000 Subject: [PATCH] fix(marketplace): poll listings recorded as pending and document where the poll runs A listing AWS gave no status for when it was created (or whose rollback cancel failed) is recorded as pending with its listing id and no attempt token. ListMarketplaceListingsToPoll only selected active and unsettled terminal rows, so such a listing never became active and a sale on it was never recorded. - ListMarketplaceListingsToPoll now also returns pending rows that hold a listing id and no client token. A pending row with a token is a list attempt in flight whose listing_id is the RI's previous listing, and is still left out. - RecordMarketplaceListingActive moves such a row to active (stamping listed_at when AWS reports a CreateDate) and the poller calls it for a pending row even without a CreateDate. - SettleMarketplaceListing also settles such a row. - Every write keeps its existing guards and re-checks the token, so a cancel or relist claim that commits while the poll waits on the row wins. The poll runs inside finalize_revocations, so docs/DEPLOYMENT.md now lists where that task is scheduled (Lambda, CloudFormation, AKS) and where it is not (Fargate, Cloud Run, GKE, Container Apps), and the schedule toggles say that turning them off also stops the poll. Superseded listings (gap 2) are not addressed: recording an earlier listing's settlement after a relist needs per-listing storage. Refs #1494 Co-Authored-By: Claude Opus 5.5 --- cloudformation/stacks/CUDly/template.yaml | 2 +- docs/DEPLOYMENT.md | 9 + internal/api/handler_marketplace_poller.go | 42 ++-- .../api/handler_marketplace_poller_test.go | 58 +++++ internal/config/interfaces.go | 18 +- ...rketplace_pending_poll_integration_test.go | 234 ++++++++++++++++++ .../store_postgres_marketplace_settlement.go | 60 +++-- ...marketplace_settlement_integration_test.go | 10 +- terraform/environments/aws/variables.tf | 2 +- .../modules/compute/aws/lambda/variables.tf | 2 +- .../modules/compute/azure/aks/variables.tf | 2 +- 11 files changed, 390 insertions(+), 49 deletions(-) create mode 100644 internal/config/store_postgres_marketplace_pending_poll_integration_test.go diff --git a/cloudformation/stacks/CUDly/template.yaml b/cloudformation/stacks/CUDly/template.yaml index 9dedf685c..c60f2a3f2 100644 --- a/cloudformation/stacks/CUDly/template.yaml +++ b/cloudformation/stacks/CUDly/template.yaml @@ -166,7 +166,7 @@ Parameters: AllowedValues: - "true" - "false" - Description: Run the finalize_revocations task on a schedule (idempotent audit repair of revoked purchases) + Description: Run the finalize_revocations task on a schedule (idempotent audit repair of revoked purchases, plus the RI Marketplace listing poll, which has no schedule of its own; disabling this also stops recording listing sales and proceeds) FinalizeRevocationsSchedule: Type: String diff --git a/docs/DEPLOYMENT.md b/docs/DEPLOYMENT.md index 1f18f2984..3cfde9765 100644 --- a/docs/DEPLOYMENT.md +++ b/docs/DEPLOYMENT.md @@ -612,6 +612,15 @@ Scheduling for AWS Fargate, CloudFormation and GCP is tracked by #712. Alternatively, use pg_cron for database-native scheduling. +### RI Marketplace Listing Poll + +CUDly learns that an RI listed on the AWS Reserved Instance Marketplace sold, closed or was canceled on AWS, and records its estimated proceeds and AWS fee, only through the marketplace listing poll. The poll has no schedule of its own: it runs inside the server's `finalize_revocations` task, after the revocation repair. It therefore runs only where `finalize_revocations` is scheduled, and turning that schedule off also stops the poll: + +- **AWS Lambda:** hourly by default. `enable_finalize_revocations_schedule` (default `true`) and `finalize_revocations_schedule` (default `rate(1 hour)`) in `terraform/environments/aws`. +- **CloudFormation:** hourly by default. The `EnableFinalizeRevocationsSchedule` (default `true`) and `FinalizeRevocationsSchedule` (default `rate(1 hour)`) stack parameters on `cloudformation/stacks/CUDly`. +- **AKS:** hourly by default, as a Kubernetes CronJob, when `deploy_kubernetes_resources` is true. `enable_finalize_revocations_schedule` (default `true`) and `finalize_revocations_schedule` (default `30 * * * *`) on the AKS module. +- **AWS Fargate, GCP Cloud Run, GKE and Azure Container Apps:** nothing schedules `finalize_revocations`, so the poll does not run and listings stay as the list and cancel endpoints recorded them. Until #712 wires the task there, call `POST /api/scheduled/finalize_revocations` (authenticated like the other `/api/scheduled/` tasks) from your own scheduler, about hourly. + --- ## Troubleshooting diff --git a/internal/api/handler_marketplace_poller.go b/internal/api/handler_marketplace_poller.go index 659ddb593..4ae65dec2 100644 --- a/internal/api/handler_marketplace_poller.go +++ b/internal/api/handler_marketplace_poller.go @@ -2,16 +2,19 @@ package api // handler_marketplace_poller.go is the RI Marketplace status poller (issue // #27). The list and cancel endpoints record a listing as active or canceled; -// nothing else told CUDly when AWS sold it. The poller reads every active -// listing from AWS (DescribeReservedInstancesListings, signed as the purchase's -// own account like the endpoints) and records the listing's AWS CreateDate -// while it is active, and its terminal state with the realized proceeds and -// the AWS fee once it closes (all sold) or is canceled after a partial sale. +// nothing else told CUDly when AWS sold it. The poller reads from AWS +// (DescribeReservedInstancesListings, signed as the purchase's own account +// like the endpoints) every active listing, and every listing recorded as +// pending because AWS gave no status when it was created (issue #1494). It +// records the listing as active with its AWS CreateDate while it is for sale, +// and its terminal state with the realized proceeds and the AWS fee once it +// closes (all sold) or is canceled after a partial sale. // -// It runs from the finalize_revocations scheduled sweep (see -// internal/server/handler.go). Its writes are the store's status-guarded -// UPDATEs, so a second run, the cancel handler or a relist racing a poll never -// double-record proceeds or undo a recorded cancel. +// It runs from the finalize_revocations scheduled task (see +// internal/server/handler_marketplace_poll.go), so only where that task is +// scheduled; docs/DEPLOYMENT.md lists the platforms. Its writes are the +// store's status-guarded UPDATEs, so a second run, the cancel handler or a +// relist racing a poll never double-record proceeds or undo a recorded cancel. import ( "context" @@ -218,12 +221,7 @@ func (h *Handler) pollMarketplaceListing(ctx context.Context, row *config.Purcha func (h *Handler) recordMarketplaceListingStatus(ctx context.Context, row *config.PurchaseHistoryRecord, st marketplaceListingStatus) (marketplacePollOutcome, error) { switch st.State { case config.ListingStateActive: - if !st.CreatedAt.IsZero() { - if _, err := h.config.RecordMarketplaceListingActive(ctx, row.PurchaseID, row.ListingID, st.CreatedAt); err != nil { - return marketplacePollSkipped, err - } - } - return marketplacePollActive, nil + return h.recordMarketplaceListingActive(ctx, row, st) case config.ListingStateClosed, config.ListingStateCancelled: settlement, priced := marketplaceSettlement(row, st) applied, err := h.config.SettleMarketplaceListing(ctx, settlement) @@ -239,3 +237,17 @@ func (h *Handler) recordMarketplaceListingStatus(ctx context.Context, row *confi } return marketplacePollSkipped, nil } + +// recordMarketplaceListingActive records a listing AWS reports for sale. An +// active row needs a write only to stamp the CreateDate. A row recorded as +// pending is moved to active even when AWS reported no CreateDate, so it can +// be canceled and relisted like any active listing (issue #1494). +func (h *Handler) recordMarketplaceListingActive(ctx context.Context, row *config.PurchaseHistoryRecord, st marketplaceListingStatus) (marketplacePollOutcome, error) { + if st.CreatedAt.IsZero() && !strings.EqualFold(row.ListingState, config.ListingStatePending) { + return marketplacePollActive, nil + } + if _, err := h.config.RecordMarketplaceListingActive(ctx, row.PurchaseID, row.ListingID, st.CreatedAt); err != nil { + return marketplacePollSkipped, err + } + return marketplacePollActive, nil +} diff --git a/internal/api/handler_marketplace_poller_test.go b/internal/api/handler_marketplace_poller_test.go index 2983c88fa..4ea7fe0de 100644 --- a/internal/api/handler_marketplace_poller_test.go +++ b/internal/api/handler_marketplace_poller_test.go @@ -3,6 +3,7 @@ package api import ( "context" "errors" + "fmt" "testing" "time" @@ -157,6 +158,63 @@ func TestPollMarketplaceListings_ActiveStampsListedAtOnly(t *testing.T) { store.AssertNotCalled(t, "SettleMarketplaceListing") } +func pendingRow(purchaseID, listingID string) config.PurchaseHistoryRecord { + row := listedRow(purchaseID, listingID) + row.ListingState = config.ListingStatePending + return row +} + +// A listing recorded as pending (AWS gave no status when it was created) that +// AWS now reports active is recorded as active, even without a CreateDate +// (issue #1494). +func TestPollMarketplaceListings_PendingRowBecomesActive(t *testing.T) { + for _, created := range []time.Time{{}, pollerCreateDate} { + t.Run(fmt.Sprintf("create date %v", !created.IsZero()), func(t *testing.T) { + store := &MockConfigStore{} + f := newPollerFixtureWithStore(t, store, pendingRow("ri-1", "ril-1")) + f.aws.statuses["ril-1"] = marketplaceListingStatus{ListingID: "ril-1", State: config.ListingStateActive, CreatedAt: created} + f.store.On("RecordMarketplaceListingActive", mock.Anything, "ri-1", "ril-1", created).Return(true, nil).Once() + + res, err := f.h.PollMarketplaceListings(context.Background()) + + require.NoError(t, err) + assert.Equal(t, MarketplacePollResult{Found: 1, Active: 1}, *res) + store.AssertExpectations(t) + store.AssertNotCalled(t, "SettleMarketplaceListing") + }) + } +} + +// An active row whose CreateDate AWS did not report needs no write. +func TestPollMarketplaceListings_ActiveRowWithoutCreateDateNeedsNoWrite(t *testing.T) { + store := &MockConfigStore{} + f := newPollerFixtureWithStore(t, store, listedRow("ri-1", "ril-1")) + f.aws.statuses["ril-1"] = marketplaceListingStatus{ListingID: "ril-1", State: config.ListingStateActive} + + res, err := f.h.PollMarketplaceListings(context.Background()) + + require.NoError(t, err) + assert.Equal(t, MarketplacePollResult{Found: 1, Active: 1}, *res) + store.AssertNotCalled(t, "RecordMarketplaceListingActive") +} + +// A pending row whose listing sold out before any poll is settled like an +// active one (issue #1494). +func TestPollMarketplaceListings_PendingRowSettles(t *testing.T) { + f := newPollerFixture(t, pendingRow("ri-1", "ril-1")) + f.aws.statuses["ril-1"] = marketplaceListingStatus{ListingID: "ril-1", State: config.ListingStateClosed, + CreatedAt: pollerCreateDate, SoldCount: 1, UnitPrice: 1000, PriceKnown: true} + settled := f.captureSettle(true) + + res, err := f.h.PollMarketplaceListings(context.Background()) + + require.NoError(t, err) + assert.Equal(t, MarketplacePollResult{Found: 1, Settled: 1}, *res) + require.Len(t, *settled, 1) + assert.Equal(t, config.ListingStateClosed, (*settled)[0].State) + assert.InDelta(t, 880, *(*settled)[0].ProceedsReceived, 1e-9) +} + func TestPollMarketplaceListings_CanceledAfterPartialSale(t *testing.T) { f := newPollerFixture(t, listedRow("ri-1", "ril-1")) f.aws.statuses["ril-1"] = marketplaceListingStatus{ListingID: "ril-1", State: config.ListingStateCancelled, diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 4046e1969..20b8abb54 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -487,20 +487,24 @@ type StoreInterface interface { ReleaseMarketplaceListingClaim(ctx context.Context, purchaseID, clientToken, priorState string, keepAttempt bool) error // ListMarketplaceListingsToPoll returns the AWS rows whose listing the - // marketplace status poller reads (issue #27): active listings, and + // marketplace status poller reads (issue #27): active listings, listings + // recorded as pending with no list attempt in flight (issue #1494), and // closed/canceled ones it has seen whose proceeds are not recorded yet. ListMarketplaceListingsToPoll(ctx context.Context) ([]PurchaseHistoryRecord, error) - // RecordMarketplaceListingActive stamps listed_at on a row whose listing - // AWS still reports active. It writes only while the row holds listingID - // in the active state and never writes listing_state. Reports whether the - // row changed. + // RecordMarketplaceListingActive records a listing AWS reports active: + // it stamps listed_at (a zero listedAt keeps the stored one) while the row + // holds listingID as active, and also moves the state to active while the + // row holds it as pending with no list attempt in flight (issue #1494). + // It never writes the state of any other row. Reports whether the row + // changed. RecordMarketplaceListingActive(ctx context.Context, purchaseID, listingID string, listedAt time.Time) (bool, error) // SettleMarketplaceListing records a listing that closed or was canceled // on AWS, with its proceeds and AWS fee when known. It applies once per - // listing, only while the row still holds it and is active or already in - // the same terminal state. Reports whether the row was written. + // listing, only while the row still holds it and is active, pending with + // no list attempt in flight, or already in the same terminal state. + // Reports whether the row was written. SettleMarketplaceListing(ctx context.Context, settlement MarketplaceListingSettlement) (bool, error) // ClaimRIExchangeIdempotencyKey atomically claims key for an RI exchange diff --git a/internal/config/store_postgres_marketplace_pending_poll_integration_test.go b/internal/config/store_postgres_marketplace_pending_poll_integration_test.go new file mode 100644 index 000000000..3be4a3562 --- /dev/null +++ b/internal/config/store_postgres_marketplace_pending_poll_integration_test.go @@ -0,0 +1,234 @@ +//go:build integration +// +build integration + +package config + +// Real-DB tests for polling a listing recorded as pending (issue #1494). A +// listing AWS gave no status for when it was created is recorded as pending +// with its listing id and no attempt token. The poller must move it to active +// or settle it, but never touch a pending row that is a list attempt in +// flight (it holds a token, and its listing_id is the RI's previous listing), +// and never undo a cancel the cancel handler recorded meanwhile. + +import ( + "context" + "fmt" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestMarketplacePendingListingPoll_Postgres(t *testing.T) { + ctx := context.Background() + conn := testhelpers.RequireMigratedDatabase(ctx, t, getTestMigrationsPath()).DB + store := NewPostgresStore(conn) + pool := conn.Pool() + + t.Run("RecordedPendingBecomesActive", func(t *testing.T) { checkPendingBecomesActive(ctx, t, store, pool) }) + t.Run("RecordedPendingSettles", func(t *testing.T) { checkPendingSettles(ctx, t, store, pool) }) + t.Run("MidClaimIsLeftAlone", func(t *testing.T) { checkMidClaimLeftAlone(ctx, t, store, pool) }) + t.Run("CancelCommittingBeforeActivationWins", func(t *testing.T) { checkCancelBeatsActivation(ctx, t, store, pool) }) + t.Run("RelistClaimCommittingBeforePollIsLeftAlone", func(t *testing.T) { checkClaimBeatsPendingPoll(ctx, t, store, pool) }) + t.Run("ParallelCancelAndPollKeepCancel", func(t *testing.T) { checkParallelPendingCancelAndPoll(ctx, t, store, pool) }) +} + +// seedMidClaim leaves id pending with an attempt token, as a marketplace-list +// request does while it creates a new listing: the row still holds its +// previous, canceled listing previousListingID. +func seedMidClaim(ctx context.Context, t *testing.T, store *PostgresStore, id, previousListingID string) { + t.Helper() + seedListing(ctx, t, store, "aws", id, previousListingID, ListingStateCancelled) + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-"+id, []byte(`[{"term_months":6,"price":40}]`)) + require.NoError(t, err) + require.NotNil(t, claim) +} + +func readClientToken(ctx context.Context, t *testing.T, pool *pgxpool.Pool, id string) string { + t.Helper() + var token string + require.NoError(t, pool.QueryRow(ctx, `SELECT COALESCE(listing_client_token, '') FROM purchase_history + WHERE purchase_id = $1`, id).Scan(&token)) + return token +} + +func checkPendingBecomesActive(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + at := listedAtFixture() + seedListing(ctx, t, store, "aws", "pending-active", "ril-pa", ListingStatePending) + + changed, err := store.RecordMarketplaceListingActive(ctx, "pending-active", "ril-pa", at) + require.NoError(t, err) + assert.True(t, changed) + got := readSettlement(ctx, t, pool, "pending-active") + assert.Equal(t, ListingStateActive, got.state) + require.NotNil(t, got.listedAt) + assert.True(t, at.Equal(*got.listedAt)) + + changed, err = store.RecordMarketplaceListingActive(ctx, "pending-active", "ril-pa", at) + require.NoError(t, err) + assert.False(t, changed, "an unchanged active row is not rewritten") + + // AWS reported no CreateDate: the state still moves, listed_at stays unset. + seedListing(ctx, t, store, "aws", "pending-active-nodate", "ril-pan", "Pending") + changed, err = store.RecordMarketplaceListingActive(ctx, "pending-active-nodate", "ril-pan", time.Time{}) + require.NoError(t, err) + assert.True(t, changed) + got = readSettlement(ctx, t, pool, "pending-active-nodate") + assert.Equal(t, ListingStateActive, got.state) + assert.Nil(t, got.listedAt) + + // A zero CreateDate never clears the listed_at an earlier poll stored. + changed, err = store.RecordMarketplaceListingActive(ctx, "pending-active", "ril-pa", time.Time{}) + require.NoError(t, err) + assert.False(t, changed) + require.NotNil(t, readSettlement(ctx, t, pool, "pending-active").listedAt) + + // A listing id the row does not hold is ignored. + seedListing(ctx, t, store, "aws", "pending-other", "ril-po", ListingStatePending) + changed, err = store.RecordMarketplaceListingActive(ctx, "pending-other", "ril-elsewhere", at) + require.NoError(t, err) + assert.False(t, changed) + assert.Equal(t, ListingStatePending, readSettlement(ctx, t, pool, "pending-other").state) +} + +func checkPendingSettles(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + seedListing(ctx, t, store, "aws", "pending-settle", "ril-ps1", ListingStatePending) + + applied, err := store.SettleMarketplaceListing(ctx, closedSettlement("pending-settle", "ril-ps1", 880, 120)) + require.NoError(t, err) + assert.True(t, applied) + got := readSettlement(ctx, t, pool, "pending-settle") + assert.Equal(t, ListingStateClosed, got.state) + require.NotNil(t, got.proceeds) + assert.InDelta(t, 880, *got.proceeds, 1e-9) + + rows, err := store.ListMarketplaceListingsToPoll(ctx) + require.NoError(t, err) + for _, r := range rows { + assert.NotEqual(t, "pending-settle", r.PurchaseID, "a settled listing is not polled again") + } +} + +func checkMidClaimLeftAlone(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + seedMidClaim(ctx, t, store, "pending-mid-claim", "ril-mc") + + changed, err := store.RecordMarketplaceListingActive(ctx, "pending-mid-claim", "ril-mc", listedAtFixture()) + require.NoError(t, err) + assert.False(t, changed) + canceledSettle := closedSettlement("pending-mid-claim", "ril-mc", 0, 0) + canceledSettle.State = ListingStateCancelled + applied, err := store.SettleMarketplaceListing(ctx, canceledSettle) + require.NoError(t, err) + assert.False(t, applied) + + got := readSettlement(ctx, t, pool, "pending-mid-claim") + assert.Equal(t, ListingStatePending, got.state) + assert.Nil(t, got.listedAt) + assert.Equal(t, "tok-pending-mid-claim", readClientToken(ctx, t, pool, "pending-mid-claim"), "the attempt token is kept") +} + +// The poller read the pending listing as active from AWS, then the cancel +// handler commits its cancel while the poller's write waits on the row. +func checkCancelBeatsActivation(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + seedListing(ctx, t, store, "aws", "pending-race-cancel", "ril-prc", ListingStatePending) + commit := holdCancel(ctx, t, pool, "pending-race-cancel", "ril-prc") + + done := make(chan guardedResult, 1) + go func() { + changed, err := store.RecordMarketplaceListingActive(ctx, "pending-race-cancel", "ril-prc", listedAtFixture()) + done <- guardedResult{changed, err} + }() + testhelpers.WaitForLockWait(ctx, t, pool, "%listed_at%") + commit() + + got := testhelpers.Await(t, done) + require.NoError(t, got.err) + assert.False(t, got.applied) + assert.Equal(t, ListingStateCancelled, readSettlement(ctx, t, pool, "pending-race-cancel").state) +} + +// The cancel handler cancels the pending listing and a list request claims the +// RI again, both committing while the poller's writes for the old listing wait +// on the row. The row is then pending again with the same listing_id, but it +// is a claim in flight: the poller must not touch it. +func checkClaimBeatsPendingPoll(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + writes := map[string]func(id string) (bool, error){ + "activate": func(id string) (bool, error) { + return store.RecordMarketplaceListingActive(ctx, id, "ril-pcl", listedAtFixture()) + }, + "settle": func(id string) (bool, error) { + return store.SettleMarketplaceListing(ctx, closedSettlement(id, "ril-pcl", 100, 10)) + }, + } + for name, write := range writes { + t.Run(name, func(t *testing.T) { + id := "pending-race-claim-" + name + seedListing(ctx, t, store, "aws", id, "ril-pcl", ListingStatePending) + tx, err := pool.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { _ = tx.Rollback(ctx) }) + _, err = tx.Exec(ctx, cancelWriteSQL, "ril-pcl", ListingStateCancelled, id) + require.NoError(t, err) + _, err = tx.Exec(ctx, `UPDATE purchase_history SET listing_state = $2, listing_client_token = $3 + WHERE purchase_id = $1`, id, ListingStatePending, "tok-"+id) + require.NoError(t, err) + + done := make(chan guardedResult, 1) + go func() { + applied, writeErr := write(id) + done <- guardedResult{applied, writeErr} + }() + testhelpers.WaitForLockWait(ctx, t, pool, "%UPDATE purchase_history%") + require.NoError(t, tx.Commit(ctx)) + + got := testhelpers.Await(t, done) + require.NoError(t, got.err) + assert.False(t, got.applied) + row := readSettlement(ctx, t, pool, id) + assert.Equal(t, ListingStatePending, row.state) + assert.Nil(t, row.proceeds) + assert.Equal(t, "tok-"+id, readClientToken(ctx, t, pool, id)) + }) + } +} + +// The cancel handler resolving a pending listing and the poller (which read it +// as active, or as closed) running at once always end with the cancel +// recorded. +func checkParallelPendingCancelAndPoll(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + for round := range 10 { + id := fmt.Sprintf("pending-parallel-%d", round) + seedListing(ctx, t, store, "aws", id, "ril-pp", ListingStatePending) + seen := MarketplaceListingAttempt{ListingID: "ril-pp", State: ListingStatePending} + + start := make(chan struct{}) + var wg sync.WaitGroup + var resolveErr, activateErr, settleErr error + var resolved bool + wg.Go(func() { + <-start + resolved, resolveErr = store.ResolveMarketplaceListingAttempt(ctx, id, seen, "ril-pp", ListingStateCancelled) + }) + wg.Go(func() { + <-start + _, activateErr = store.RecordMarketplaceListingActive(ctx, id, "ril-pp", listedAtFixture()) + }) + wg.Go(func() { + <-start + _, settleErr = store.SettleMarketplaceListing(ctx, closedSettlement(id, "ril-pp", 100, 10)) + }) + close(start) + wg.Wait() + require.NoError(t, resolveErr) + require.NoError(t, activateErr) + require.NoError(t, settleErr) + assert.True(t, resolved, "round %d: the cancel is always recorded", round) + row := readSettlement(ctx, t, pool, id) + assert.Equal(t, ListingStateCancelled, row.state, "round %d", round) + assert.Equal(t, "ril-pp", row.listingID, "round %d", round) + } +} diff --git a/internal/config/store_postgres_marketplace_settlement.go b/internal/config/store_postgres_marketplace_settlement.go index 756ca9637..e9cbac6f8 100644 --- a/internal/config/store_postgres_marketplace_settlement.go +++ b/internal/config/store_postgres_marketplace_settlement.go @@ -15,11 +15,15 @@ import ( ) // ListMarketplaceListingsToPoll returns the AWS purchase_history rows whose -// listing the poller must read: every active listing, plus closed or canceled -// listings the poller has seen (listed_at set) whose proceeds are not recorded -// yet. The second set covers a cancel recorded by the cancel handler before the -// poller saw the listing end, and a terminal listing whose sale price could -// not be determined on an earlier poll. Only the fields the poller uses are +// listing the poller must read: every active listing; every listing recorded +// as pending, i.e. one AWS gave no status for when it was created, or whose +// rollback cancel failed (issue #1494); and closed or canceled listings the +// poller has seen (listed_at set) whose proceeds are not recorded yet. A +// pending row that holds a list attempt's token is a claim in flight, whose +// listing_id is the RI's previous listing, so it is left out. The last set +// covers a cancel recorded by the cancel handler before the poller saw the +// listing end, and a terminal listing whose sale price could not be +// determined on an earlier poll. Only the fields the poller uses are // populated: AccountID, PurchaseID, Provider, Region, CloudAccountID, // ListingID and ListingState. One row is returned per (purchase_id, // listing_id), the earliest, so legacy duplicate rows cost one AWS call. @@ -32,12 +36,14 @@ func (s *PostgresStore) ListMarketplaceListingsToPoll(ctx context.Context) ([]Pu WHERE provider = 'aws' AND COALESCE(listing_id, '') <> '' AND (lower(COALESCE(listing_state, '')) = $1 + OR (lower(COALESCE(listing_state, '')) = $4 + AND listing_client_token IS NULL) OR (lower(COALESCE(listing_state, '')) IN ($2, $3) AND listed_at IS NOT NULL AND listing_proceeds_received IS NULL)) ORDER BY purchase_id, listing_id, created_at, id ` - rows, err := s.db.Query(ctx, query, ListingStateActive, ListingStateClosed, ListingStateCancelled) + rows, err := s.db.Query(ctx, query, ListingStateActive, ListingStateClosed, ListingStateCancelled, ListingStatePending) if err != nil { return nil, fmt.Errorf("failed to list marketplace listings to poll: %w", err) } @@ -60,23 +66,35 @@ func (s *PostgresStore) ListMarketplaceListingsToPoll(ctx context.Context) ([]Pu return out, nil } -// RecordMarketplaceListingActive stamps listed_at on a row whose listing AWS -// still reports active. It writes only while the row holds listingID in the -// active state, and never writes listing_state, so a cancel or relist recorded -// after the poller read AWS is kept. Any settlement still on the row belongs to -// an earlier listing of the same RI and is cleared. It reports whether the row +// RecordMarketplaceListingActive records that AWS reports the row's listing +// active. On a row that holds listingID as active it stamps listed_at and never +// writes listing_state, so a cancel or relist recorded after the poller read +// AWS is kept. On a row that holds listingID as pending with no list attempt in +// flight (no client token) it also moves the state to active (issue #1494); a +// pending row with a token is a claim whose listing_id is the RI's previous +// listing and is left alone. A zero listedAt (AWS reported no CreateDate) +// keeps the stored listed_at. Any settlement still on the row belongs to an +// earlier listing of the same RI and is cleared. It reports whether the row // changed. func (s *PostgresStore) RecordMarketplaceListingActive(ctx context.Context, purchaseID, listingID string, listedAt time.Time) (bool, error) { + var at *time.Time + if !listedAt.IsZero() { + utc := listedAt.UTC() + at = &utc + } query := ` UPDATE purchase_history - SET listed_at = $3, listing_proceeds_received = NULL, listing_fee_paid = NULL + SET listing_state = CASE WHEN lower(COALESCE(listing_state, '')) = $5 THEN $4 ELSE listing_state END, + listed_at = COALESCE($3::timestamptz, listed_at), + listing_proceeds_received = NULL, listing_fee_paid = NULL WHERE purchase_id = $1 AND listing_id = $2 - AND lower(COALESCE(listing_state, '')) = $4 - AND (listed_at IS DISTINCT FROM $3 - OR listing_proceeds_received IS NOT NULL - OR listing_fee_paid IS NOT NULL) + AND ((lower(COALESCE(listing_state, '')) = $5 AND listing_client_token IS NULL) + OR (lower(COALESCE(listing_state, '')) = $4 + AND (listed_at IS DISTINCT FROM COALESCE($3::timestamptz, listed_at) + OR listing_proceeds_received IS NOT NULL + OR listing_fee_paid IS NOT NULL))) ` - tag, err := s.db.Exec(ctx, query, purchaseID, listingID, listedAt.UTC(), ListingStateActive) + tag, err := s.db.Exec(ctx, query, purchaseID, listingID, at, ListingStateActive, ListingStatePending) if err != nil { return false, fmt.Errorf("failed to record active marketplace listing %s for purchase %s: %w", listingID, purchaseID, err) } @@ -86,7 +104,8 @@ func (s *PostgresStore) RecordMarketplaceListingActive(ctx context.Context, purc // SettleMarketplaceListing records a listing that ended on AWS: its terminal // state, listed_at and, when known, the proceeds and AWS fee. It applies only // while the row still holds the listing, has no proceeds recorded, and is -// active or already in the same terminal state. So proceeds are recorded once +// active, pending with no list attempt in flight (issue #1494), or already in +// the same terminal state. So proceeds are recorded once // however many polls race, a recorded cancel is never turned into a sale (or // the reverse), and a row that was relisted is left alone. The list paths // (UpdatePurchaseHistoryListing, ResolveMarketplaceListingAttempt) clear the @@ -114,10 +133,11 @@ func (s *PostgresStore) SettleMarketplaceListing(ctx context.Context, st Marketp listing_fee_paid = $6 WHERE purchase_id = $1 AND listing_id = $2 AND listing_proceeds_received IS NULL - AND lower(COALESCE(listing_state, '')) IN ($7, $3) + AND (lower(COALESCE(listing_state, '')) IN ($7, $3) + OR (lower(COALESCE(listing_state, '')) = $8 AND listing_client_token IS NULL)) ` tag, err := s.db.Exec(ctx, query, st.PurchaseID, st.ListingID, state, listedAt, - st.ProceedsReceived, st.FeePaid, ListingStateActive) + st.ProceedsReceived, st.FeePaid, ListingStateActive, ListingStatePending) if err != nil { return false, fmt.Errorf("failed to settle marketplace listing %s for purchase %s: %w", st.ListingID, st.PurchaseID, err) } diff --git a/internal/config/store_postgres_marketplace_settlement_integration_test.go b/internal/config/store_postgres_marketplace_settlement_integration_test.go index 85739974f..ca8c1c89a 100644 --- a/internal/config/store_postgres_marketplace_settlement_integration_test.go +++ b/internal/config/store_postgres_marketplace_settlement_integration_test.go @@ -107,11 +107,13 @@ func checkPollSelection(ctx context.Context, t *testing.T, store *PostgresStore, setSettlement(ctx, t, pool, "poll-cancel-unsettled", &seen, nil, nil) seedListing(ctx, t, store, "aws", "poll-closed-unpriced", "ril-cl", ListingStateClosed) setSettlement(ctx, t, pool, "poll-closed-unpriced", &seen, nil, nil) + // A listing recorded in AWS's own pending state (issue #1494). + seedListing(ctx, t, store, "aws", "poll-pending", "ril-pending", ListingStatePending) // Excluded: settled, never seen by the poller, mid-claim, not listed, not AWS. seedListing(ctx, t, store, "aws", "poll-closed-settled", "ril-cs", ListingStateClosed) setSettlement(ctx, t, pool, "poll-closed-settled", &seen, settleAmount(88), settleAmount(12)) seedListing(ctx, t, store, "aws", "poll-cancel-legacy", "ril-legacy", ListingStateCancelled) - seedListing(ctx, t, store, "aws", "poll-pending", "ril-pending", ListingStatePending) + seedMidClaim(ctx, t, store, "poll-mid-claim", "ril-mid-claim") seedListing(ctx, t, store, "aws", "poll-unlisted", "", "") seedListing(ctx, t, store, "azure", "poll-azure", "ril-azure", ListingStateActive) @@ -123,7 +125,9 @@ func checkPollSelection(ctx context.Context, t *testing.T, store *PostgresStore, got[r.PurchaseID] = r } } - assert.ElementsMatch(t, []string{"poll-active", "poll-active-upper", "poll-cancel-unsettled", "poll-closed-unpriced"}, slices.Collect(maps.Keys(got))) + assert.ElementsMatch(t, []string{"poll-active", "poll-active-upper", "poll-cancel-unsettled", "poll-closed-unpriced", "poll-pending"}, + slices.Collect(maps.Keys(got))) + assert.Equal(t, ListingStatePending, got["poll-pending"].ListingState) active := got["poll-active"] assert.Equal(t, "ril-active", active.ListingID) assert.Equal(t, ListingStateActive, active.ListingState) @@ -217,7 +221,7 @@ func checkSettleGuards(ctx context.Context, t *testing.T, store *PostgresStore, applied, err = store.SettleMarketplaceListing(ctx, closedSettlement("settle-relisted", "ril-old", 5, 1)) require.NoError(t, err) assert.False(t, applied) - seedListing(ctx, t, store, "aws", "settle-pending", "ril-sp", ListingStatePending) + seedMidClaim(ctx, t, store, "settle-pending", "ril-sp") applied, err = store.SettleMarketplaceListing(ctx, closedSettlement("settle-pending", "ril-sp", 5, 1)) require.NoError(t, err) assert.False(t, applied) diff --git a/terraform/environments/aws/variables.tf b/terraform/environments/aws/variables.tf index e2170b3ca..015e60436 100644 --- a/terraform/environments/aws/variables.tf +++ b/terraform/environments/aws/variables.tf @@ -431,7 +431,7 @@ variable "send_notifications_schedule" { } variable "enable_finalize_revocations_schedule" { - description = "Enable the scheduled finalize_revocations sweep (idempotent audit repair)." + description = "Enable the scheduled finalize_revocations sweep (idempotent audit repair, plus the RI Marketplace listing poll, which has no schedule of its own: disabling this also stops recording listing sales and proceeds)." type = bool default = true } diff --git a/terraform/modules/compute/aws/lambda/variables.tf b/terraform/modules/compute/aws/lambda/variables.tf index 39122eeda..79c87e20a 100644 --- a/terraform/modules/compute/aws/lambda/variables.tf +++ b/terraform/modules/compute/aws/lambda/variables.tf @@ -280,7 +280,7 @@ variable "send_notifications_schedule" { } variable "enable_finalize_revocations_schedule" { - description = "Enable the scheduled finalize_revocations sweep (idempotent audit repair of revoked purchases)." + description = "Enable the scheduled finalize_revocations sweep (idempotent audit repair of revoked purchases, plus the RI Marketplace listing poll, which has no schedule of its own: disabling this also stops recording listing sales and proceeds)." type = bool default = true } diff --git a/terraform/modules/compute/azure/aks/variables.tf b/terraform/modules/compute/azure/aks/variables.tf index 881703f53..0f235a6a7 100644 --- a/terraform/modules/compute/azure/aks/variables.tf +++ b/terraform/modules/compute/azure/aks/variables.tf @@ -313,7 +313,7 @@ variable "send_notifications_schedule" { } variable "enable_finalize_revocations_schedule" { - description = "Schedule the finalize_revocations sweep (idempotent audit repair of revoked purchases)." + description = "Schedule the finalize_revocations sweep (idempotent audit repair of revoked purchases, plus the RI Marketplace listing poll, which has no schedule of its own: disabling this also stops recording listing sales and proceeds)." type = bool default = true }