From d7aa121f4d273a0ed3899bd32c741ea745da0e91 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Wed, 28 Oct 2026 21:22:38 +0000 Subject: [PATCH 1/3] fix(marketplace): keep each RI Marketplace listing's settlement in its own row purchase_history describes only an RI's latest marketplace listing, and a relist (or the claim that starts one) moves the row to the new listing on purpose (#1289). A listing that sold partially and was canceled before a poll saw the sale was then never polled again, and its proceeds were lost. - Migration 000153 adds marketplace_listing_settlements, one row per AWS listing id with the purchase_history row it was recorded on, the RI id, status, listed_at, proceeds, fee, currency and timestamps. It backfills one row per listing purchase_history records a settlement for. - Trigger purchase_history_keep_superseded_listing copies a listing into the table when the row stops describing it: its listing_id changes, or a list claim stores a token over it. ON CONFLICT DO NOTHING keeps a row the poller already wrote. - ListMarketplaceListingsToPoll also returns those kept listings, under the same rules as purchase_history rows, once per listing. - SettleMarketplaceListing writes, in one transaction, the purchase_history row while it still holds the listing (as before, so its columns stay the latest listing's settlement) and then the listing's own row, created or filled only while its amounts are unset. purchase_history is locked first, the same order a relist takes, so the two cannot deadlock. - GetMarketplaceListingProceedsTotals sums an RI's proceeds and fees over its listings from the table alone, so a sale is never counted twice. Closes #1494 Co-Authored-By: Claude Opus 5.5 --- internal/api/handler_marketplace_poller.go | 4 +- internal/config/interfaces.go | 24 +- internal/config/store_postgres.go | 9 +- .../store_postgres_marketplace_attempt.go | 5 +- ...ce_listing_settlements_integration_test.go | 409 ++++++++++++++++++ .../store_postgres_marketplace_settlement.go | 262 +++++++++-- internal/config/types.go | 18 + ...3_marketplace_listing_settlements.down.sql | 8 + ...153_marketplace_listing_settlements.up.sql | 132 ++++++ ...53_marketplace_listing_settlements_test.go | 177 ++++++++ internal/mocks/stores.go | 15 + 11 files changed, 1003 insertions(+), 60 deletions(-) create mode 100644 internal/config/store_postgres_marketplace_listing_settlements_integration_test.go create mode 100644 internal/database/postgres/migrations/000153_marketplace_listing_settlements.down.sql create mode 100644 internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql create mode 100644 internal/database/postgres/migrations/000153_marketplace_listing_settlements_test.go diff --git a/internal/api/handler_marketplace_poller.go b/internal/api/handler_marketplace_poller.go index 4ae65dec2..d171081ef 100644 --- a/internal/api/handler_marketplace_poller.go +++ b/internal/api/handler_marketplace_poller.go @@ -8,7 +8,9 @@ package api // 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. +// closes (all sold) or is canceled after a partial sale. A listing the RI was +// relisted from is still read and settled, from the per-listing settlement +// rows the store keeps (issue #1494). // // It runs from the finalize_revocations scheduled task (see // internal/server/handler_marketplace_poll.go), so only where that task is diff --git a/internal/config/interfaces.go b/internal/config/interfaces.go index 8105a1643..f34129c33 100644 --- a/internal/config/interfaces.go +++ b/internal/config/interfaces.go @@ -514,10 +514,12 @@ type StoreInterface interface { // they are cleared because AWS created nothing reusable. 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, 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 returns the AWS listings the 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, + // both the RI's current listing and the listings it was relisted from + // (issue #1494). ListMarketplaceListingsToPoll(ctx context.Context) ([]PurchaseHistoryRecord, error) // RecordMarketplaceListingActive records a listing AWS reports active: @@ -529,12 +531,18 @@ type StoreInterface interface { 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, pending with - // no list attempt in flight, or already in the same terminal state. - // Reports whether the row was written. + // on AWS, with its proceeds and AWS fee when known, in the listing's own + // settlement row and, while the RI still holds the listing, on its + // purchase_history row (issue #1494). It applies once per listing, only + // while the listing is active, pending with no list attempt in flight, or + // already in the same terminal state. Reports whether a row was written. SettleMarketplaceListing(ctx context.Context, settlement MarketplaceListingSettlement) (bool, error) + // GetMarketplaceListingProceedsTotals sums each RI's marketplace proceeds + // and fees over all of its listings (issue #1494), one total per purchase + // and currency. RIs with no recorded listing are left out. + GetMarketplaceListingProceedsTotals(ctx context.Context, purchaseIDs []string) ([]MarketplaceListingProceedsTotal, error) + // ClaimRIExchangeIdempotencyKey atomically claims key for an RI exchange // submit, so a client that retries a timed-out execute request cannot // commit the same exchange twice (issue #1642). key is a fingerprint of diff --git a/internal/config/store_postgres.go b/internal/config/store_postgres.go index af0fce80e..cd617adff 100644 --- a/internal/config/store_postgres.go +++ b/internal/config/store_postgres.go @@ -2429,6 +2429,9 @@ func (s *PostgresStore) UpdatePurchaseHistoryListing(ctx context.Context, purcha // Writing a new listing id (a relist) drops the previous listing's // listed_at and settlement in the same statement (issue #27), so the new // listing can be settled even if it ends before the poller sees it active. + // The previous listing itself is kept in marketplace_listing_settlements + // by trigger purchase_history_keep_superseded_listing (migration 000153, + // issue #1494), where the poller still settles it. query := ` UPDATE purchase_history SET listing_id = $1, listing_state = $2, @@ -2473,7 +2476,11 @@ func (s *PostgresStore) StampOfferingClass(ctx context.Context, purchaseID, offe // priceSchedule in the same statement unless an earlier attempt already stored a // token, and returns the row's values as of the claim (issue #525): the handler // must not trust a row read before the claim. It returns (nil, nil) when the row -// could not be claimed (active, pending without a token, or absent). +// could not be claimed (active, pending without a token, or absent). The +// claimed row keeps the previous listing id but no longer records that +// listing's state, so trigger purchase_history_keep_superseded_listing copies +// the previous listing into marketplace_listing_settlements when the token is +// first stored (migration 000153, issue #1494). // // Rows saved before migration 000108 can share a purchase_id (#267). The claim // then works on the earliest row (created_at, then id), the one diff --git a/internal/config/store_postgres_marketplace_attempt.go b/internal/config/store_postgres_marketplace_attempt.go index 895fa53ef..798e9256c 100644 --- a/internal/config/store_postgres_marketplace_attempt.go +++ b/internal/config/store_postgres_marketplace_attempt.go @@ -57,8 +57,9 @@ func (s *PostgresStore) GetMarketplaceListingAttempt(ctx context.Context, purcha // caller just canceled). state must be an ended state (canceled or closed) or // empty, which clears a stuck pending row that has no listing; an empty // listingID clears the listing id too. A new listing id drops the previous -// listing's listed_at and settlement (issue #27). It reports whether the row -// was written. +// listing's listed_at and settlement (issue #27); the previous listing is +// kept in marketplace_listing_settlements (migration 000153, issue #1494). It +// reports whether the row was written. func (s *PostgresStore) ResolveMarketplaceListingAttempt(ctx context.Context, purchaseID string, seen MarketplaceListingAttempt, listingID, state string) (bool, error) { if state != "" && !isEndedListingState(state) { return false, fmt.Errorf("cannot resolve marketplace listing %s for purchase %s to live state %q", listingID, purchaseID, state) diff --git a/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go b/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go new file mode 100644 index 000000000..6dc36d812 --- /dev/null +++ b/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go @@ -0,0 +1,409 @@ +//go:build integration +// +build integration + +package config + +// Real-DB tests for per-listing RI Marketplace settlements (issue #1494, gap +// 2). purchase_history only describes an RI's latest listing, so a listing +// that sold partially and was canceled, then relisted before a poll saw the +// sale, used to lose its proceeds. Migration 000153 keeps every listing in +// marketplace_listing_settlements; these tests check that both listings' +// proceeds are kept, that polls are idempotent and race-safe, and that a sale +// is never counted twice. + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type listingSettlementRow struct { + purchaseID, status string + listedAt *time.Time + proceeds, fee *float64 + settledAt *time.Time +} + +// readListingSettlement returns the marketplace_listing_settlements row of +// listingID, or nil when there is none. +func readListingSettlement(ctx context.Context, t *testing.T, pool *pgxpool.Pool, listingID string) *listingSettlementRow { + t.Helper() + var r listingSettlementRow + err := pool.QueryRow(ctx, ` + SELECT purchase_id, status, listed_at, proceeds_received::float8, fee_paid::float8, settled_at + FROM marketplace_listing_settlements WHERE listing_id = $1`, listingID). + Scan(&r.purchaseID, &r.status, &r.listedAt, &r.proceeds, &r.fee, &r.settledAt) + if errors.Is(err, pgx.ErrNoRows) { + return nil + } + require.NoError(t, err) + return &r +} + +// requireListingProceeds asserts the recorded proceeds of listingID's +// settlement row. +func requireListingProceeds(ctx context.Context, t *testing.T, pool *pgxpool.Pool, listingID string, want float64) { + t.Helper() + row := readListingSettlement(ctx, t, pool, listingID) + require.NotNil(t, row, "listing %s has a settlement row", listingID) + require.NotNil(t, row.proceeds, "listing %s has proceeds", listingID) + assert.InDelta(t, want, *row.proceeds, 1e-9, "listing %s proceeds", listingID) +} + +// requireRowProceeds asserts the proceeds on purchase id's purchase_history row. +func requireRowProceeds(ctx context.Context, t *testing.T, pool *pgxpool.Pool, id string, want float64) { + t.Helper() + got := readSettlement(ctx, t, pool, id).proceeds + require.NotNil(t, got, "purchase %s has proceeds", id) + assert.InDelta(t, want, *got, 1e-9, "purchase %s proceeds", id) +} + +// relist runs the marketplace-list handler's store calls for a new listing: +// claim the slot, then record the listing AWS created. +func relist(ctx context.Context, t *testing.T, store *PostgresStore, id, newListingID string) { + t.Helper() + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-"+newListingID, []byte(`[{"term_months":6,"price":40}]`)) + require.NoError(t, err) + require.NotNil(t, claim) + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, newListingID, ListingStateActive)) +} + +// seedCanceledSeenListing records listingID on a new RI as a listing the poller +// saw active (listed_at set) and the cancel handler then canceled: the state a +// partial sale is left in until the next poll. +func seedCanceledSeenListing(ctx context.Context, t *testing.T, store *PostgresStore, id, listingID string) { + t.Helper() + seedListing(ctx, t, store, "aws", id, listingID, ListingStateActive) + _, err := store.RecordMarketplaceListingActive(ctx, id, listingID, listedAtFixture()) + require.NoError(t, err) + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, listingID, ListingStateCancelled)) +} + +func canceledSettlement(id, listingID string, proceeds, fee float64) MarketplaceListingSettlement { + st := closedSettlement(id, listingID, proceeds, fee) + st.State = ListingStateCancelled + return st +} + +// pollListings returns the poll rows of purchase id, keyed by listing id. +func pollListings(ctx context.Context, t *testing.T, store *PostgresStore, id string) map[string]PurchaseHistoryRecord { + t.Helper() + rows, err := store.ListMarketplaceListingsToPoll(ctx) + require.NoError(t, err) + out := map[string]PurchaseHistoryRecord{} + for _, r := range rows { + if r.PurchaseID == id { + _, dup := out[r.ListingID] + require.False(t, dup, "listing %s returned twice", r.ListingID) + out[r.ListingID] = r + } + } + return out +} + +// requireTotal asserts the RI's proceeds totals across listings. +func requireTotal(ctx context.Context, t *testing.T, store *PostgresStore, id string, proceeds, fee float64, settled, unsettled int) { + t.Helper() + totals, err := store.GetMarketplaceListingProceedsTotals(ctx, []string{id}) + require.NoError(t, err) + require.Len(t, totals, 1) + got := totals[0] + assert.Equal(t, id, got.PurchaseID) + assert.Equal(t, "USD", got.Currency) + assert.InDelta(t, proceeds, got.ProceedsReceived, 1e-9, "proceeds total") + assert.InDelta(t, fee, got.FeePaid, 1e-9, "fee total") + assert.Equal(t, settled, got.SettledListings, "settled listings") + assert.Equal(t, unsettled, got.UnsettledListings, "unsettled listings") +} + +func TestMarketplaceListingSettlements_Postgres(t *testing.T) { + ctx := context.Background() + conn := testhelpers.RequireMigratedDatabase(ctx, t, getTestMigrationsPath()).DB + store := NewPostgresStore(conn) + pool := conn.Pool() + + t.Run("RelistAfterPartialSaleKeepsBothListings", func(t *testing.T) { checkRelistKeepsBothListings(ctx, t, store, pool) }) + t.Run("RepollIsIdempotent", func(t *testing.T) { checkRepollIdempotent(ctx, t, store, pool) }) + t.Run("UnpricedSupersededListingIsPolledUntilPriced", func(t *testing.T) { checkUnpricedSuperseded(ctx, t, store, pool) }) + t.Run("ReleasedClaimCountsListingOnce", func(t *testing.T) { checkReleasedClaimCountsOnce(ctx, t, store, pool) }) + t.Run("SettleWaitingOnClaimSettlesKeptListing", func(t *testing.T) { checkSettleWaitingOnClaim(ctx, t, store, pool) }) + t.Run("ConcurrentPollsSettleSupersededOnce", func(t *testing.T) { checkConcurrentSupersededSettles(ctx, t, store, pool) }) + t.Run("ConcurrentRelistAndPollKeepProceedsOnce", func(t *testing.T) { checkConcurrentRelistAndPoll(ctx, t, store, pool) }) + t.Run("TotalsOfUnknownPurchases", func(t *testing.T) { checkTotalsOfUnknownPurchases(ctx, t, store) }) +} + +// The reported bug: listing A sells partially, is canceled, and the RI is +// relisted as B before a poll sees A's sale. Both listings' proceeds must be +// recorded, and the RI's total is their sum. +func checkRelistKeepsBothListings(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "mls-relist" + seedCanceledSeenListing(ctx, t, store, id, "ril-mls-a") + relist(ctx, t, store, id, "ril-mls-b") + + due := pollListings(ctx, t, store, id) + require.Contains(t, due, "ril-mls-a", "the listing the RI was relisted from is still polled") + require.Contains(t, due, "ril-mls-b") + a := due["ril-mls-a"] + assert.Equal(t, ListingStateCancelled, a.ListingState, "the kept listing has its own last state, not the claim's") + assert.Equal(t, "acct", a.AccountID) + assert.Equal(t, "us-east-1", a.Region) + assert.Equal(t, "aws", a.Provider) + + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mls-a", 88, 12)) + require.NoError(t, err) + assert.True(t, applied, "the superseded listing's partial sale is recorded") + applied, err = store.SettleMarketplaceListing(ctx, closedSettlement(id, "ril-mls-b", 176, 24)) + require.NoError(t, err) + assert.True(t, applied) + + rowA := readListingSettlement(ctx, t, pool, "ril-mls-a") + require.NotNil(t, rowA) + assert.Equal(t, id, rowA.purchaseID) + assert.Equal(t, ListingStateCancelled, rowA.status) + require.NotNil(t, rowA.listedAt) + assert.True(t, listedAtFixture().Equal(*rowA.listedAt)) + require.NotNil(t, rowA.proceeds) + require.NotNil(t, rowA.fee) + assert.InDelta(t, 88, *rowA.proceeds, 1e-9) + assert.InDelta(t, 12, *rowA.fee, 1e-9) + assert.NotNil(t, rowA.settledAt) + rowB := readListingSettlement(ctx, t, pool, "ril-mls-b") + require.NotNil(t, rowB) + assert.Equal(t, ListingStateClosed, rowB.status) + requireListingProceeds(ctx, t, pool, "ril-mls-b", 176) + + // purchase_history keeps the latest listing's settlement. + ph := readSettlement(ctx, t, pool, id) + assert.Equal(t, "ril-mls-b", ph.listingID) + assert.Equal(t, ListingStateClosed, ph.state) + requireRowProceeds(ctx, t, pool, id, 176) + + requireTotal(ctx, t, store, id, 264, 36, 2, 0) + assert.Empty(t, pollListings(ctx, t, store, id), "settled listings are not polled again") +} + +func checkRepollIdempotent(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "mls-repoll" + seedCanceledSeenListing(ctx, t, store, id, "ril-mrp-a") + relist(ctx, t, store, id, "ril-mrp-b") + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mrp-a", 88, 12)) + require.NoError(t, err) + require.True(t, applied) + applied, err = store.SettleMarketplaceListing(ctx, closedSettlement(id, "ril-mrp-b", 50, 7)) + require.NoError(t, err) + require.True(t, applied) + + for range 3 { + applied, err = store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mrp-a", 1000, 1)) + require.NoError(t, err) + assert.False(t, applied, "a superseded listing is settled once") + applied, err = store.SettleMarketplaceListing(ctx, closedSettlement(id, "ril-mrp-b", 1000, 1)) + require.NoError(t, err) + assert.False(t, applied, "the current listing is settled once") + } + requireListingProceeds(ctx, t, pool, "ril-mrp-a", 88) + requireListingProceeds(ctx, t, pool, "ril-mrp-b", 50) + requireRowProceeds(ctx, t, pool, id, 50) + requireTotal(ctx, t, store, id, 138, 19, 2, 0) +} + +// A superseded listing whose sale price is not known yet records its state +// and stays due; a later poll that prices it records it once. +func checkUnpricedSuperseded(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "mls-unpriced" + seedCanceledSeenListing(ctx, t, store, id, "ril-mup-a") + relist(ctx, t, store, id, "ril-mup-b") + + at := listedAtFixture() + unpriced := MarketplaceListingSettlement{PurchaseID: id, ListingID: "ril-mup-a", State: ListingStateCancelled, ListedAt: &at} + applied, err := store.SettleMarketplaceListing(ctx, unpriced) + require.NoError(t, err) + assert.True(t, applied) + row := readListingSettlement(ctx, t, pool, "ril-mup-a") + require.NotNil(t, row) + assert.Nil(t, row.proceeds) + assert.Nil(t, row.settledAt) + assert.Contains(t, pollListings(ctx, t, store, id), "ril-mup-a", "an unpriced listing is read again") + requireTotal(ctx, t, store, id, 0, 0, 0, 1) + + applied, err = store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mup-a", 44, 6)) + require.NoError(t, err) + assert.True(t, applied) + assert.NotContains(t, pollListings(ctx, t, store, id), "ril-mup-a") + requireTotal(ctx, t, store, id, 44, 6, 1, 0) +} + +// A failed relist releases its claim, so the RI's row holds listing A again +// while A is also kept in the settlement table. A is polled once per run and +// its proceeds count once. +func checkReleasedClaimCountsOnce(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "mls-release" + seedCanceledSeenListing(ctx, t, store, id, "ril-mrl-a") + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-mrl", []byte(`[{"term_months":6,"price":40}]`)) + require.NoError(t, err) + require.NotNil(t, claim) + require.NotNil(t, readListingSettlement(ctx, t, pool, "ril-mrl-a"), "the claim keeps the listing it replaces") + require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "tok-mrl", claim.PriorState, false)) + + due := pollListings(ctx, t, store, id) + require.Len(t, due, 1) + require.Contains(t, due, "ril-mrl-a") + + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mrl-a", 88, 12)) + require.NoError(t, err) + assert.True(t, applied) + requireRowProceeds(ctx, t, pool, id, 88) + requireListingProceeds(ctx, t, pool, "ril-mrl-a", 88) + requireTotal(ctx, t, store, id, 88, 12, 1, 0) + + applied, err = store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mrl-a", 88, 12)) + require.NoError(t, err) + assert.False(t, applied) + requireTotal(ctx, t, store, id, 88, 12, 1, 0) +} + +// A poll of listing A that queues behind a relist claim committing on the +// same RI re-checks the row, finds the claim, and records A's sale in the row +// the claim kept for it instead of dropping it. +func checkSettleWaitingOnClaim(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "mls-held-claim" + seedCanceledSeenListing(ctx, t, store, id, "ril-mhc-a") + tx, err := pool.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { _ = tx.Rollback(ctx) }) + _, err = tx.Exec(ctx, `UPDATE purchase_history SET listing_state = 'pending', + listing_client_token = 'tok-mhc' WHERE purchase_id = $1`, id) + require.NoError(t, err) + + done := make(chan guardedResult, 1) + go func() { + applied, settleErr := store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mhc-a", 88, 12)) + done <- guardedResult{applied, settleErr} + }() + testhelpers.WaitForLockWait(ctx, t, pool, "%listing_proceeds_received%") + require.NoError(t, tx.Commit(ctx)) + + got := testhelpers.Await(t, done) + require.NoError(t, got.err) + assert.True(t, got.applied) + ph := readSettlement(ctx, t, pool, id) + assert.Equal(t, ListingStatePending, ph.state, "the claim is left alone") + assert.Nil(t, ph.proceeds) + requireListingProceeds(ctx, t, pool, "ril-mhc-a", 88) + requireTotal(ctx, t, store, id, 88, 12, 1, 0) +} + +// Many pollers settling the same superseded listing at once record it exactly +// once, with the winner's values. +func checkConcurrentSupersededSettles(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const pollers = 8 + for round := range 5 { + id := fmt.Sprintf("mls-parallel-%d", round) + listingA := fmt.Sprintf("ril-mpp-a-%d", round) + seedCanceledSeenListing(ctx, t, store, id, listingA) + relist(ctx, t, store, id, fmt.Sprintf("ril-mpp-b-%d", round)) + + start := make(chan struct{}) + results := make([]guardedResult, pollers) + var wg sync.WaitGroup + for i := range pollers { + wg.Go(func() { + <-start + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, listingA, float64(100+i), float64(i))) + results[i] = guardedResult{applied, err} + }) + } + close(start) + wg.Wait() + + winners, winner := 0, 0 + for i, r := range results { + require.NoError(t, r.err) + if r.applied { + winners++ + winner = i + } + } + require.Equal(t, 1, winners, "round %d: exactly one settle applies", round) + requireListingProceeds(ctx, t, pool, listingA, float64(100+winner)) + requireTotal(ctx, t, store, id, float64(100+winner), float64(winner), 1, 0) + } +} + +// A relist and two polls of the listing it replaces run at once and end with +// that listing's sale recorded exactly once, and no deadlock. In even rounds +// all three start together; in odd rounds the polls start once the relist's +// claim has committed (so the claim has kept the listing) and race the write +// of the new listing id. Both ways the listing's row can be created are hit: +// by a poll that settles it while the RI still holds it, and by the claim. +func checkConcurrentRelistAndPoll(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + keptByClaim := 0 + for round := range 20 { + id := fmt.Sprintf("mls-race-%d", round) + listingA := fmt.Sprintf("ril-mrr-a-%d", round) + seedCanceledSeenListing(ctx, t, store, id, listingA) + + start, claimed := make(chan struct{}), make(chan struct{}) + var wg sync.WaitGroup + var relistErr error + wg.Go(func() { + <-start + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-"+id, []byte(`[{"term_months":6,"price":40}]`)) + close(claimed) + if err != nil || claim == nil { + relistErr = fmt.Errorf("claim: %v, %w", claim, err) + return + } + relistErr = store.UpdatePurchaseHistoryListing(ctx, id, fmt.Sprintf("ril-mrr-b-%d", round), ListingStateActive) + }) + results := make([]guardedResult, 2) + for i := range results { + wg.Go(func() { + <-start + if round%2 == 1 { + <-claimed + } + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, listingA, 88, 12)) + results[i] = guardedResult{applied, err} + }) + } + close(start) + wg.Wait() + require.NoError(t, relistErr, "round %d", round) + require.NoError(t, results[0].err, "round %d", round) + require.NoError(t, results[1].err, "round %d", round) + assert.True(t, results[0].applied != results[1].applied, "round %d: exactly one poll records the sale", round) + + row := readListingSettlement(ctx, t, pool, listingA) + require.NotNil(t, row, "round %d", round) + require.NotNil(t, row.proceeds, "round %d", round) + assert.Equal(t, ListingStateCancelled, row.status, "round %d", round) + requireTotal(ctx, t, store, id, 88, 12, 1, 0) + var claimFirst bool + require.NoError(t, pool.QueryRow(ctx, `SELECT settled_at > created_at FROM marketplace_listing_settlements + WHERE listing_id = $1`, listingA).Scan(&claimFirst)) + if claimFirst { + keptByClaim++ + } + } + t.Logf("listing row created by the relist claim in %d of 20 rounds", keptByClaim) + assert.GreaterOrEqual(t, keptByClaim, 10, "every odd round settles a listing the claim kept") +} + +func checkTotalsOfUnknownPurchases(ctx context.Context, t *testing.T, store *PostgresStore) { + totals, err := store.GetMarketplaceListingProceedsTotals(ctx, nil) + require.NoError(t, err) + assert.Empty(t, totals) + totals, err = store.GetMarketplaceListingProceedsTotals(ctx, []string{"mls-never-listed"}) + require.NoError(t, err) + assert.Empty(t, totals) +} diff --git a/internal/config/store_postgres_marketplace_settlement.go b/internal/config/store_postgres_marketplace_settlement.go index e9cbac6f8..2555e2fe9 100644 --- a/internal/config/store_postgres_marketplace_settlement.go +++ b/internal/config/store_postgres_marketplace_settlement.go @@ -1,37 +1,43 @@ package config // Store side of the RI Marketplace status poller (issue #27). The poller reads -// each listing from AWS and records what it finds with the two status-guarded -// UPDATEs below. The guards, not the poller's copy of the row, decide whether +// each listing from AWS and records what it finds with the status-guarded +// writes below. The guards, not the poller's copy of the row, decide whether // a write applies, so two poller runs, or a poll racing the cancel handler or a // relist, never record proceeds twice, revive a canceled listing, or write to a // listing the row no longer holds. +// +// Every listing's settlement is also kept in its own marketplace_listing_settlements +// row (migration 000153, issue #1494). purchase_history only describes an RI's +// latest listing, so a listing the RI was relisted from is polled and settled +// from that table instead of being lost. import ( "context" "fmt" "strings" "time" + + "github.com/jackc/pgx/v5" ) -// ListMarketplaceListingsToPoll returns the AWS purchase_history rows whose -// 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. -func (s *PostgresStore) ListMarketplaceListingsToPoll(ctx context.Context) ([]PurchaseHistoryRecord, error) { - query := ` - SELECT DISTINCT ON (purchase_id, listing_id) - account_id, purchase_id, provider, region, cloud_account_id::text, - listing_id, listing_state +// marketplaceListingsToPollSQL selects the listings the poller must read, one +// row per (purchase_id, listing_id). $1..$4 are ListingStateActive, +// ListingStateClosed, ListingStateCancelled and ListingStatePending. +// +// The first branch is the RI's current listing on purchase_history. The +// second is a listing the RI has since moved on from (marketplace_listing_settlements, +// issue #1494), selected by the same rules; its account, region and cloud +// account come from the purchase_history row it was made from. When both +// branches name a listing the purchase_history one is returned (src 0), so the +// poller sees the row's live state. +const marketplaceListingsToPollSQL = ` + SELECT DISTINCT ON (purchase_id, listing_id) + account_id, purchase_id, provider, region, cloud_account_id::text, + listing_id, listing_state + FROM ( + SELECT 0 AS src, created_at, id, account_id, purchase_id, provider, region, + cloud_account_id, listing_id, listing_state FROM purchase_history WHERE provider = 'aws' AND COALESCE(listing_id, '') <> '' @@ -41,9 +47,36 @@ func (s *PostgresStore) ListMarketplaceListingsToPoll(ctx context.Context) ([]Pu 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, ListingStatePending) + UNION ALL + SELECT 1, p.created_at, p.id, p.account_id, s.purchase_id, p.provider, p.region, + p.cloud_account_id, s.listing_id, s.status + FROM marketplace_listing_settlements s + JOIN purchase_history p ON p.id = s.purchase_history_id + WHERE p.provider = 'aws' + AND s.proceeds_received IS NULL + AND (lower(s.status) IN ($1, $4) + OR (lower(s.status) IN ($2, $3) AND s.listed_at IS NOT NULL)) + ) due + ORDER BY purchase_id, listing_id, src, created_at, id +` + +// ListMarketplaceListingsToPoll returns the AWS listings 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. It reads both the RI's current listing +// on purchase_history and the listings the RI was relisted from, which +// marketplace_listing_settlements keeps (issue #1494). A pending row that +// holds a list attempt's token is a claim in flight, whose listing_id is the +// RI's previous listing; that listing is polled from the settlement table. The +// terminal 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. +func (s *PostgresStore) ListMarketplaceListingsToPoll(ctx context.Context) ([]PurchaseHistoryRecord, error) { + rows, err := s.db.Query(ctx, marketplaceListingsToPollSQL, ListingStateActive, ListingStateClosed, ListingStateCancelled, ListingStatePending) if err != nil { return nil, fmt.Errorf("failed to list marketplace listings to poll: %w", err) } @@ -101,45 +134,178 @@ func (s *PostgresStore) RecordMarketplaceListingActive(ctx context.Context, purc return tag.RowsAffected() > 0, nil } +// settleCurrentListingSQL records a terminal listing on the purchase_history +// row(s) that still hold it. It applies only while the row has no proceeds +// recorded and is active, pending with no list attempt in flight (issue +// #1494), or already in the same terminal state. Arguments are those of +// settleListingArgs. +const settleCurrentListingSQL = ` + UPDATE purchase_history + SET listing_state = $3, + listed_at = COALESCE($4::timestamptz, listed_at), + listing_proceeds_received = $5, + 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) + OR (lower(COALESCE(listing_state, '')) = $8 AND listing_client_token IS NULL)) +` + +// upsertListingSettlementSQL records the same settlement in the listing's own +// row, creating it from the purchase_history row settleCurrentListingSQL just +// wrote (the earliest, for legacy duplicates). An existing row is only filled +// while its amounts are unset and its status is active, pending or the same +// terminal state, and only for the same RI, so a listing's proceeds are +// recorded once. +const upsertListingSettlementSQL = ` + INSERT INTO marketplace_listing_settlements AS s ( + listing_id, purchase_history_id, purchase_id, status, listed_at, + proceeds_received, fee_paid, settled_at) + SELECT p.listing_id, p.id, p.purchase_id, $3, COALESCE($4::timestamptz, p.listed_at), + $5, $6, CASE WHEN $5::numeric IS NOT NULL THEN now() END + FROM purchase_history p + WHERE p.purchase_id = $1 AND p.listing_id = $2 + ORDER BY p.created_at, p.id + LIMIT 1 + ON CONFLICT (listing_id) DO UPDATE + SET status = EXCLUDED.status, + listed_at = COALESCE(EXCLUDED.listed_at, s.listed_at), + proceeds_received = EXCLUDED.proceeds_received, + fee_paid = EXCLUDED.fee_paid, + settled_at = EXCLUDED.settled_at, + updated_at = now() + WHERE s.purchase_id = EXCLUDED.purchase_id + AND s.proceeds_received IS NULL + AND lower(s.status) IN ($7, $8, $3) +` + +// settleSupersededListingSQL records the settlement of a listing the RI no +// longer holds on purchase_history, in the row the relist kept for it, under +// the same guards as upsertListingSettlementSQL. +const settleSupersededListingSQL = ` + UPDATE marketplace_listing_settlements AS s + SET status = $3, + listed_at = COALESCE($4::timestamptz, s.listed_at), + proceeds_received = $5, + fee_paid = $6, + settled_at = CASE WHEN $5::numeric IS NOT NULL THEN now() END, + updated_at = now() + WHERE s.listing_id = $2 AND s.purchase_id = $1 + AND s.proceeds_received IS NULL + AND lower(s.status) IN ($7, $8, $3) +` + // 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, 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 -// previous listing's settlement when they write a new listing id, so a relist -// settles here even if it ends before any poll sees it active. It reports -// whether the row was written. +// state, listed_at and, when known, the proceeds and AWS fee. In one +// transaction it writes the purchase_history row while the row still holds the +// listing (the RI's latest listing settlement, as before) and the listing's +// marketplace_listing_settlements row (issue #1494), which is what per-RI +// totals read. Each write applies only while that row has no proceeds +// recorded and is active, pending with no list attempt in flight, 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 +// listing the RI was relisted from is still settled, in its own row. The +// purchase_history row is written first, so a poll and a relist of the same +// RI take their locks in the same order. It reports whether either row was +// written. func (s *PostgresStore) SettleMarketplaceListing(ctx context.Context, st MarketplaceListingSettlement) (bool, error) { + args, err := settleListingArgs(st) + if err != nil { + return false, err + } + var applied bool + err = s.WithTx(ctx, func(tx pgx.Tx) error { + var txErr error + applied, txErr = settleMarketplaceListingTx(ctx, tx, args) + return txErr + }) + if err != nil { + return false, fmt.Errorf("failed to settle marketplace listing %s for purchase %s: %w", st.ListingID, st.PurchaseID, err) + } + return applied, nil +} + +// settleListingArgs validates st and returns the arguments the settle +// statements share: purchase id, listing id, lowercase terminal state, +// listed_at (UTC, nil to keep), proceeds, fee, then ListingStateActive and +// ListingStatePending. +func settleListingArgs(st MarketplaceListingSettlement) ([]any, error) { state := strings.ToLower(st.State) if state != ListingStateClosed && state != ListingStateCancelled { - return false, fmt.Errorf("cannot settle marketplace listing %s in state %q", st.ListingID, st.State) + return nil, fmt.Errorf("cannot settle marketplace listing %s in state %q", st.ListingID, st.State) } if (st.ProceedsReceived == nil) != (st.FeePaid == nil) { - return false, fmt.Errorf("marketplace listing %s: proceeds and fee must be recorded together", st.ListingID) + return nil, fmt.Errorf("marketplace listing %s: proceeds and fee must be recorded together", st.ListingID) } var listedAt *time.Time if st.ListedAt != nil { utc := st.ListedAt.UTC() listedAt = &utc } + return []any{st.PurchaseID, st.ListingID, state, listedAt, st.ProceedsReceived, st.FeePaid, + ListingStateActive, ListingStatePending}, nil +} + +// settleMarketplaceListingTx runs SettleMarketplaceListing's writes in tx. A +// listing the purchase_history row still holds gets its settlement row +// created or filled; one the row no longer holds can only be filled in the +// settlement row its relist kept. +func settleMarketplaceListingTx(ctx context.Context, tx pgx.Tx, args []any) (bool, error) { + tag, err := tx.Exec(ctx, settleCurrentListingSQL, args...) + if err != nil { + return false, err + } + rowApplied := tag.RowsAffected() > 0 + listingSQL := settleSupersededListingSQL + if rowApplied { + listingSQL = upsertListingSettlementSQL + } + tag, err = tx.Exec(ctx, listingSQL, args...) + if err != nil { + return false, err + } + return rowApplied || tag.RowsAffected() > 0, nil +} + +// GetMarketplaceListingProceedsTotals returns, for each of purchaseIDs that +// has any marketplace listing recorded, the net proceeds and AWS fees summed +// over all of its settled listings, from marketplace_listing_settlements only +// (issue #1494). Each listing has one row there, so a sale is counted once +// however often the RI was relisted; purchase_history's settlement columns +// (the latest listing only) are never added in. One total is returned per +// purchase and currency, ordered by purchase id then currency. +func (s *PostgresStore) GetMarketplaceListingProceedsTotals(ctx context.Context, purchaseIDs []string) ([]MarketplaceListingProceedsTotal, error) { + if len(purchaseIDs) == 0 { + return nil, nil + } query := ` - UPDATE purchase_history - SET listing_state = $3, - listed_at = COALESCE($4::timestamptz, listed_at), - listing_proceeds_received = $5, - 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) - OR (lower(COALESCE(listing_state, '')) = $8 AND listing_client_token IS NULL)) + SELECT purchase_id, currency, + COALESCE(SUM(proceeds_received), 0)::float8, + COALESCE(SUM(fee_paid), 0)::float8, + count(proceeds_received), + count(*) - count(proceeds_received) + FROM marketplace_listing_settlements + WHERE purchase_id = ANY($1) + GROUP BY purchase_id, currency + ORDER BY purchase_id, currency ` - tag, err := s.db.Exec(ctx, query, st.PurchaseID, st.ListingID, state, listedAt, - st.ProceedsReceived, st.FeePaid, ListingStateActive, ListingStatePending) + rows, err := s.db.Query(ctx, query, purchaseIDs) if err != nil { - return false, fmt.Errorf("failed to settle marketplace listing %s for purchase %s: %w", st.ListingID, st.PurchaseID, err) + return nil, fmt.Errorf("failed to read marketplace listing proceeds totals: %w", err) } - return tag.RowsAffected() > 0, nil + defer rows.Close() + + var out []MarketplaceListingProceedsTotal + for rows.Next() { + var t MarketplaceListingProceedsTotal + if err := rows.Scan(&t.PurchaseID, &t.Currency, &t.ProceedsReceived, &t.FeePaid, + &t.SettledListings, &t.UnsettledListings); err != nil { + return nil, fmt.Errorf("failed to scan marketplace listing proceeds total: %w", err) + } + out = append(out, t) + } + if err := rows.Err(); err != nil { + return nil, fmt.Errorf("failed to read marketplace listing proceeds totals: %w", err) + } + return out, nil } diff --git a/internal/config/types.go b/internal/config/types.go index 5772656ec..44a8cb09a 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -1287,6 +1287,24 @@ type MarketplaceListingSettlement struct { FeePaid *float64 } +// MarketplaceListingProceedsTotal is one RI's RI Marketplace sales summed over +// every listing it has had (issue #1494), in one currency. +type MarketplaceListingProceedsTotal struct { + // PurchaseID is the RI (reservation) id. + PurchaseID string + // Currency is the ISO code of the amounts; the poller records USD only. + Currency string + // ProceedsReceived is the seller's net proceeds and FeePaid the AWS fee, + // summed over the settled listings. + ProceedsReceived float64 + FeePaid float64 + // SettledListings counts the listings whose amounts are recorded, and + // UnsettledListings those still without amounts (not ended yet, ended + // with an unknown price, or ended before the poller saw them). + SettledListings int + UnsettledListings int +} + // AWS EC2 ReservedInstancesListing status values, mirroring the // ec2types.ListingStatus enum. Stored verbatim in purchase_history.listing_state // so the strings must match AWS exactly. ListingStatePending is additionally diff --git a/internal/database/postgres/migrations/000153_marketplace_listing_settlements.down.sql b/internal/database/postgres/migrations/000153_marketplace_listing_settlements.down.sql new file mode 100644 index 000000000..d1291c2c0 --- /dev/null +++ b/internal/database/postgres/migrations/000153_marketplace_listing_settlements.down.sql @@ -0,0 +1,8 @@ +-- Revert migration 000153 (issue #1494): drop the per-listing settlement table +-- and the trigger that fills it. This restores the 000152 schema. +-- purchase_history's settlement columns were not changed by 000153 and keep +-- each RI's latest listing settlement. Lossy by design: settlements of +-- superseded listings exist only in this table and are dropped with it. +DROP TRIGGER IF EXISTS purchase_history_keep_superseded_listing ON purchase_history; +DROP FUNCTION IF EXISTS keep_superseded_marketplace_listing(); +DROP TABLE IF EXISTS marketplace_listing_settlements; diff --git a/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql b/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql new file mode 100644 index 000000000..d47d662b3 --- /dev/null +++ b/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql @@ -0,0 +1,132 @@ +-- Migration 000153: one settlement row per RI Marketplace listing (issue #1494). +-- +-- purchase_history holds one listing per RI: listing_id, listing_state, +-- listed_at, listing_proceeds_received and listing_fee_paid describe the RI's +-- latest listing, and a relist (or the claim that starts one) moves the row on +-- to the new listing on purpose (issue #27). A listing that sold partially and +-- was canceled before a poll saw the sale therefore lost its proceeds when the +-- RI was relisted: the row no longer named it, so nothing polled it again. +-- +-- marketplace_listing_settlements keeps every listing's settlement in its own +-- row, keyed by the AWS listing id (AWS listing ids are unique, and each one +-- belongs to exactly one RI): +-- +-- purchase_history_id the purchase_history row the listing was recorded on +-- (for legacy duplicate rows of one RI, the earliest +-- when the poller or the backfill creates the row, the +-- first row updated when the trigger does) +-- purchase_id the RI (reservation) id, for per-RI totals +-- status the listing's state as last recorded (lowercase AWS +-- ListingStatus) +-- listed_at the listing's AWS CreateDate, once the poller saw it +-- proceeds_received seller's net USD for the sold instances, NULL until +-- fee_paid settled; the AWS fee on that sale. Set together. +-- currency ISO currency of the two amounts (the poller only +-- prices USD tiers) +-- settled_at when the amounts were recorded +-- +-- The reference cascades: a purchase_history row that is deleted takes its +-- listings' rows with it, as it takes its own settlement columns. No code path +-- deletes purchase_history rows today. +-- +-- Writers: +-- * The marketplace poller (SettleMarketplaceListing) upserts the row of a +-- listing it read as closed or canceled. The upsert only fills a row whose +-- amounts are still NULL, so concurrent or repeated polls record a +-- listing's proceeds once. +-- * Trigger purchase_history_keep_superseded_listing copies a listing into +-- this table, as purchase_history last recorded it, at the moment the row +-- stops describing it: when listing_id changes, or when a list attempt's +-- claim sets listing_client_token (the claimed row keeps the previous +-- listing id but its state becomes the attempt's). ON CONFLICT DO NOTHING +-- keeps a row the poller already wrote. The poller keeps polling such a +-- listing from here until it is settled, with the same rules it uses for +-- purchase_history rows. +-- +-- purchase_history's settlement columns keep their meaning: the latest +-- listing's settlement, which the poller still writes in the same transaction. +-- Per-RI totals across listings are read from this table only, never added to +-- purchase_history's columns, so no listing is counted twice. +-- +-- Backfill: one row per listing that purchase_history currently records a +-- settlement for (listing_proceeds_received IS NOT NULL). Legacy duplicate +-- rows of one RI hold the same listing, so the earliest row is used. +-- +-- Read-only detection query: listings purchase_history records a settlement +-- for that have no row here (expected to return nothing after this migration): +-- +-- SELECT p.purchase_id, p.listing_id, p.listing_proceeds_received +-- FROM purchase_history p +-- WHERE p.listing_proceeds_received IS NOT NULL +-- AND COALESCE(p.listing_id, '') <> '' +-- AND NOT EXISTS (SELECT 1 FROM marketplace_listing_settlements s +-- WHERE s.listing_id = p.listing_id); +-- +-- Down migration: drops the trigger, its function and this table, which +-- restores the 000152 schema. purchase_history's columns were never changed, +-- so they still hold each RI's latest listing settlement; the settlements of +-- superseded listings exist only here and are lost on rollback. + +CREATE TABLE IF NOT EXISTS marketplace_listing_settlements ( + listing_id TEXT PRIMARY KEY CHECK (listing_id <> ''), + purchase_history_id UUID NOT NULL REFERENCES purchase_history (id) ON DELETE CASCADE, + purchase_id VARCHAR(255) NOT NULL, + status TEXT NOT NULL, + listed_at TIMESTAMPTZ, + proceeds_received NUMERIC(12,2), + fee_paid NUMERIC(12,2), + currency CHAR(3) NOT NULL DEFAULT 'USD' CHECK (currency ~ '^[A-Z]{3}$'), + settled_at TIMESTAMPTZ, + created_at TIMESTAMPTZ NOT NULL DEFAULT now(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT now(), + CONSTRAINT marketplace_listing_settlements_amounts_check + CHECK ((proceeds_received IS NULL) = (fee_paid IS NULL) + AND (proceeds_received IS NULL) = (settled_at IS NULL)) +); + +CREATE INDEX IF NOT EXISTS idx_marketplace_listing_settlements_purchase + ON marketplace_listing_settlements (purchase_id); + +CREATE INDEX IF NOT EXISTS idx_marketplace_listing_settlements_history_row + ON marketplace_listing_settlements (purchase_history_id); + +INSERT INTO marketplace_listing_settlements ( + listing_id, purchase_history_id, purchase_id, status, listed_at, + proceeds_received, fee_paid, settled_at +) +SELECT DISTINCT ON (listing_id) + listing_id, id, purchase_id, lower(COALESCE(listing_state, '')), listed_at, + listing_proceeds_received, COALESCE(listing_fee_paid, 0), now() +FROM purchase_history +WHERE listing_proceeds_received IS NOT NULL + AND COALESCE(listing_id, '') <> '' +ORDER BY listing_id, created_at, id +ON CONFLICT (listing_id) DO NOTHING; + +CREATE OR REPLACE FUNCTION keep_superseded_marketplace_listing() +RETURNS TRIGGER AS $$ +BEGIN + INSERT INTO marketplace_listing_settlements ( + listing_id, purchase_history_id, purchase_id, status, listed_at, + proceeds_received, fee_paid, settled_at + ) VALUES ( + OLD.listing_id, OLD.id, OLD.purchase_id, lower(OLD.listing_state), OLD.listed_at, + OLD.listing_proceeds_received, + CASE WHEN OLD.listing_proceeds_received IS NOT NULL THEN COALESCE(OLD.listing_fee_paid, 0) END, + CASE WHEN OLD.listing_proceeds_received IS NOT NULL THEN now() END + ) + ON CONFLICT (listing_id) DO NOTHING; + RETURN NULL; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS purchase_history_keep_superseded_listing ON purchase_history; +CREATE TRIGGER purchase_history_keep_superseded_listing + AFTER UPDATE OF listing_id, listing_client_token ON purchase_history + FOR EACH ROW + WHEN (COALESCE(OLD.listing_id, '') <> '' + AND COALESCE(OLD.listing_state, '') <> '' + AND OLD.listing_client_token IS NULL + AND (NEW.listing_id IS DISTINCT FROM OLD.listing_id + OR NEW.listing_client_token IS NOT NULL)) + EXECUTE FUNCTION keep_superseded_marketplace_listing(); diff --git a/internal/database/postgres/migrations/000153_marketplace_listing_settlements_test.go b/internal/database/postgres/migrations/000153_marketplace_listing_settlements_test.go new file mode 100644 index 000000000..27ed5489a --- /dev/null +++ b/internal/database/postgres/migrations/000153_marketplace_listing_settlements_test.go @@ -0,0 +1,177 @@ +//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/jackc/pgx/v5/pgxpool" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const ( + migration153Version = 153 + // listingStateCancelled is AWS's ListingStatus spelling, stored verbatim. + listingStateCancelled = "cancelled" //nolint:misspell // AWS ListingStatus enum literal, not prose +) + +// listingSettlementsTableExists reports whether migration 000153's table exists. +func listingSettlementsTableExists(ctx context.Context, t *testing.T, pool *pgxpool.Pool) bool { + t.Helper() + var exists bool + require.NoError(t, pool.QueryRow(ctx, + `SELECT to_regclass('public.marketplace_listing_settlements') IS NOT NULL`).Scan(&exists)) + return exists +} + +// seedListingRow inserts a purchase_history row with a listing and, when +// proceeds is non-nil, its settlement. createdAt orders legacy duplicates. +func seedListingRow(ctx context.Context, t *testing.T, pool *pgxpool.Pool, purchaseID, listingID, state string, + createdAt time.Time, listedAt *time.Time, proceeds, fee *float64) string { + t.Helper() + var id string + require.NoError(t, pool.QueryRow(ctx, ` + INSERT INTO purchase_history ( + account_id, purchase_id, timestamp, provider, service, region, resource_type, + term, payment, offering_class, listing_id, listing_state, created_at, + listed_at, listing_proceeds_received, listing_fee_paid + ) VALUES ('test-account', $1, $2, 'aws', 'ec2', 'us-east-1', 'm5.large', + 1, 'all-upfront', 'standard', NULLIF($3, ''), NULLIF($4, ''), $2, $5, $6, $7) + RETURNING id::text`, + purchaseID, createdAt, listingID, state, listedAt, proceeds, fee).Scan(&id)) + return id +} + +type migratedListingSettlement struct { + historyID, purchaseID, status string + listedAt *time.Time + proceeds, fee *float64 + settledAt *time.Time +} + +func readMigratedListingSettlements(ctx context.Context, t *testing.T, pool *pgxpool.Pool) map[string]migratedListingSettlement { + t.Helper() + rows, err := pool.Query(ctx, ` + SELECT listing_id, purchase_history_id::text, purchase_id, status, listed_at, + proceeds_received::float8, fee_paid::float8, settled_at + FROM marketplace_listing_settlements`) + require.NoError(t, err) + defer rows.Close() + out := map[string]migratedListingSettlement{} + for rows.Next() { + var listingID string + var r migratedListingSettlement + require.NoError(t, rows.Scan(&listingID, &r.historyID, &r.purchaseID, &r.status, &r.listedAt, + &r.proceeds, &r.fee, &r.settledAt)) + out[listingID] = r + } + require.NoError(t, rows.Err()) + return out +} + +func amount(v float64) *float64 { return &v } + +// TestMigration153_MarketplaceListingSettlements locks down migration 000153 +// (issue #1494): it creates the per-listing settlement table, backfills one row +// per listing purchase_history records a settlement for (the earliest row of +// legacy duplicates), keeps a listing the RI moves on from via its trigger, +// and its rollback drops the table and trigger while purchase_history keeps +// its settlement columns and values. +// +// Fail-before: at the previous version the table does not exist and a relist +// keeps nothing of the previous listing. +func TestMigration153_MarketplaceListingSettlements(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Minute) + defer cancel() + + container := testhelpers.RequirePostgresContainer(ctx, t) + defer container.Cleanup(ctx) + pool := container.DB.Pool() + migrationsPath := getMigrationsPath() + + prev := previousMigrationVersion(t, migration153Version) + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, prev)) + require.False(t, listingSettlementsTableExists(ctx, t, pool)) + + listedAt := time.Date(2026, 9, 1, 12, 30, 0, 0, time.UTC) + old := time.Date(2020, 1, 1, 0, 0, 0, 0, time.UTC) + settledID := seedListingRow(ctx, t, pool, "ri-153-settled", "ril-153-s", "closed", old, &listedAt, amount(880), amount(120)) + // Legacy duplicates of one RI hold the same listing: one backfilled row, + // from the earliest. + dupFirst := seedListingRow(ctx, t, pool, "ri-153-dup", "ril-153-d", listingStateCancelled, old, &listedAt, amount(88), amount(12)) + seedListingRow(ctx, t, pool, "ri-153-dup", "ril-153-d", listingStateCancelled, old.Add(time.Hour), &listedAt, amount(88), amount(12)) + // Not backfilled: unsettled, never listed. + seedListingRow(ctx, t, pool, "ri-153-unsettled", "ril-153-u", listingStateCancelled, old, &listedAt, nil, nil) + seedListingRow(ctx, t, pool, "ri-153-unlisted", "", "", old, nil, nil, nil) + + // Before 000153 a relist leaves no trace of the previous listing. + _, err := pool.Exec(ctx, `UPDATE purchase_history SET listing_id = 'ril-153-u2', listing_state = 'active' + WHERE purchase_id = 'ri-153-unsettled'`) + require.NoError(t, err) + _, err = pool.Exec(ctx, `UPDATE purchase_history SET listing_id = 'ril-153-u', listing_state = 'cancelled' + WHERE purchase_id = 'ri-153-unsettled'`) + require.NoError(t, err) + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, migration153Version)) + require.True(t, listingSettlementsTableExists(ctx, t, pool)) + got := readMigratedListingSettlements(ctx, t, pool) + require.Len(t, got, 2, "one row per settled listing") + s := got["ril-153-s"] + assert.Equal(t, settledID, s.historyID) + assert.Equal(t, "ri-153-settled", s.purchaseID) + assert.Equal(t, "closed", s.status) + require.NotNil(t, s.listedAt) + assert.True(t, listedAt.Equal(*s.listedAt)) + require.NotNil(t, s.proceeds) + require.NotNil(t, s.fee) + assert.InDelta(t, 880, *s.proceeds, 1e-9) + assert.InDelta(t, 120, *s.fee, 1e-9) + assert.NotNil(t, s.settledAt) + d := got["ril-153-d"] + assert.Equal(t, dupFirst, d.historyID, "legacy duplicates backfill from the earliest row") + assert.InDelta(t, 88, *d.proceeds, 1e-9) + + // The trigger keeps the listing the RI moves on from, as it was recorded. + _, err = pool.Exec(ctx, `UPDATE purchase_history SET listing_id = 'ril-153-u2', listing_state = 'active' + WHERE purchase_id = 'ri-153-unsettled'`) + require.NoError(t, err) + got = readMigratedListingSettlements(ctx, t, pool) + require.Contains(t, got, "ril-153-u") + u := got["ril-153-u"] + assert.Equal(t, listingStateCancelled, u.status) + require.NotNil(t, u.listedAt) + assert.True(t, listedAt.Equal(*u.listedAt)) + assert.Nil(t, u.proceeds) + assert.Nil(t, u.settledAt) + + // Proceeds and fee are stored together, to the cent. + _, err = pool.Exec(ctx, `UPDATE marketplace_listing_settlements SET proceeds_received = 1 WHERE listing_id = 'ril-153-u'`) + require.Error(t, err, "proceeds without a fee is rejected") + + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, prev)) + require.False(t, listingSettlementsTableExists(ctx, t, pool)) + var triggers, functions int + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM pg_trigger + WHERE tgname = 'purchase_history_keep_superseded_listing'`).Scan(&triggers)) + require.NoError(t, pool.QueryRow(ctx, `SELECT count(*) FROM pg_proc + WHERE proname = 'keep_superseded_marketplace_listing'`).Scan(&functions)) + assert.Zero(t, triggers) + assert.Zero(t, functions) + var phProceeds *float64 + require.NoError(t, pool.QueryRow(ctx, `SELECT listing_proceeds_received::float8 FROM purchase_history + WHERE purchase_id = 'ri-153-settled'`).Scan(&phProceeds)) + require.NotNil(t, phProceeds, "the rollback keeps purchase_history's settlement") + assert.InDelta(t, 880, *phProceeds, 1e-9) + _, err = pool.Exec(ctx, `UPDATE purchase_history SET listing_id = 'ril-153-u3' WHERE purchase_id = 'ri-153-unsettled'`) + require.NoError(t, err, "a relist after the rollback needs no trigger") + + // Re-applying after a rollback backfills again. + require.NoError(t, migrations.MigrateToVersion(ctx, pool, migrationsPath, migration153Version)) + assert.Len(t, readMigratedListingSettlements(ctx, t, pool), 2) +} diff --git a/internal/mocks/stores.go b/internal/mocks/stores.go index 234510ef6..ce62f622f 100644 --- a/internal/mocks/stores.go +++ b/internal/mocks/stores.go @@ -1031,6 +1031,21 @@ func (m *MockConfigStore) SettleMarketplaceListing(ctx context.Context, settleme return args.Bool(0), args.Error(1) } +// GetMarketplaceListingProceedsTotals mocks the per-RI marketplace proceeds totals (issue #1494). +func (m *MockConfigStore) GetMarketplaceListingProceedsTotals(ctx context.Context, purchaseIDs []string) ([]config.MarketplaceListingProceedsTotal, error) { + m.record("GetMarketplaceListingProceedsTotals", ctx, purchaseIDs) + args := m.Called(ctx, purchaseIDs) + var totals []config.MarketplaceListingProceedsTotal + if v := args.Get(0); v != nil { + t, ok := v.([]config.MarketplaceListingProceedsTotal) + if !ok { + m.reporter.wrongReturnType("GetMarketplaceListingProceedsTotals", "[]config.MarketplaceListingProceedsTotal", v) + } + totals = t + } + return totals, args.Error(1) +} + // ClaimRIExchangeIdempotencyKey mocks the atomic RI exchange submit claim // (issue #1642). With no registered expectation it defaults to "this call won // the claim" so the many pre-existing execute-path tests, whose subject is a From 40f2b91f02477c0930ae378361cbbc6639b8bb85 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Wed, 28 Oct 2026 21:51:10 +0000 Subject: [PATCH 2/3] fix(marketplace): leave a listing held by a list claim alone in the settlement table too The per-listing settlement write ignored #1508's rule that a row mid-claim (pending with a list attempt's token) is left alone: a poll of the listing a claim had just kept was recorded in marketplace_listing_settlements while the claim was in flight, so SettleMarketplaceListing reported applied. - SettleMarketplaceListing share-locks the purchase_history rows that hold the listing after its own UPDATE and writes no settlement row while one of them is a claim in flight. The lock makes a claim that starts meanwhile wait, so nothing is written mid-claim. The listing is settled once the claim resolves (it is then polled from the kept row) or is released. - ListMarketplaceListingsToPoll leaves such a listing out of the settlement table branch as well. - The trigger also keeps the listing when a resumed claim (a released attempt whose token was kept, back in the listing's own state) makes the row pending again, and never copies a row that is already mid-claim. Refs #1494 Co-Authored-By: Claude Opus 5.5 --- ...ce_listing_settlements_integration_test.go | 98 +++++++++++++++++-- .../store_postgres_marketplace_settlement.go | 50 ++++++++-- ...153_marketplace_listing_settlements.up.sql | 17 ++-- 3 files changed, 141 insertions(+), 24 deletions(-) diff --git a/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go b/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go index 6dc36d812..4604c3599 100644 --- a/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go +++ b/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go @@ -135,7 +135,8 @@ func TestMarketplaceListingSettlements_Postgres(t *testing.T) { t.Run("RepollIsIdempotent", func(t *testing.T) { checkRepollIdempotent(ctx, t, store, pool) }) t.Run("UnpricedSupersededListingIsPolledUntilPriced", func(t *testing.T) { checkUnpricedSuperseded(ctx, t, store, pool) }) t.Run("ReleasedClaimCountsListingOnce", func(t *testing.T) { checkReleasedClaimCountsOnce(ctx, t, store, pool) }) - t.Run("SettleWaitingOnClaimSettlesKeptListing", func(t *testing.T) { checkSettleWaitingOnClaim(ctx, t, store, pool) }) + t.Run("SettleWaitingOnClaimLeavesListingForLater", func(t *testing.T) { checkSettleWaitingOnClaim(ctx, t, store, pool) }) + t.Run("ResumedClaimKeepsListing", func(t *testing.T) { checkResumedClaimKeepsListing(ctx, t, store, pool) }) t.Run("ConcurrentPollsSettleSupersededOnce", func(t *testing.T) { checkConcurrentSupersededSettles(ctx, t, store, pool) }) t.Run("ConcurrentRelistAndPollKeepProceedsOnce", func(t *testing.T) { checkConcurrentRelistAndPoll(ctx, t, store, pool) }) t.Run("TotalsOfUnknownPurchases", func(t *testing.T) { checkTotalsOfUnknownPurchases(ctx, t, store) }) @@ -252,6 +253,7 @@ func checkReleasedClaimCountsOnce(ctx context.Context, t *testing.T, store *Post require.NoError(t, err) require.NotNil(t, claim) require.NotNil(t, readListingSettlement(ctx, t, pool, "ril-mrl-a"), "the claim keeps the listing it replaces") + requireMidClaimLeftAlone(ctx, t, store, pool, id, "ril-mrl-a") require.NoError(t, store.ReleaseMarketplaceListingClaim(ctx, id, "tok-mrl", claim.PriorState, false)) due := pollListings(ctx, t, store, id) @@ -271,9 +273,25 @@ func checkReleasedClaimCountsOnce(ctx context.Context, t *testing.T, store *Post requireTotal(ctx, t, store, id, 88, 12, 1, 0) } +// requireMidClaimLeftAlone asserts that while purchase id's row holds listingID +// as a list claim in flight, the listing is not polled and a poll's settle of +// it writes neither the row nor the listing's settlement row (#1508). +func requireMidClaimLeftAlone(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool, id, listingID string) { + t.Helper() + assert.NotContains(t, pollListings(ctx, t, store, id), listingID, "a listing held by a claim in flight is not polled") + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, listingID, 88, 12)) + require.NoError(t, err) + assert.False(t, applied, "a claim in flight is left alone") + assert.Nil(t, readSettlement(ctx, t, pool, id).proceeds) + if row := readListingSettlement(ctx, t, pool, listingID); row != nil { + assert.Nil(t, row.proceeds, "the kept listing is not settled while the claim is in flight") + } +} + // A poll of listing A that queues behind a relist claim committing on the -// same RI re-checks the row, finds the claim, and records A's sale in the row -// the claim kept for it instead of dropping it. +// same RI re-checks the row, finds the claim, and writes nothing. A is not +// lost: once the claim resolves to the new listing, A is polled from the row +// the claim kept for it and its sale is recorded once. func checkSettleWaitingOnClaim(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { const id = "mls-held-claim" seedCanceledSeenListing(ctx, t, store, id, "ril-mhc-a") @@ -294,14 +312,61 @@ func checkSettleWaitingOnClaim(ctx context.Context, t *testing.T, store *Postgre got := testhelpers.Await(t, done) require.NoError(t, got.err) - assert.True(t, got.applied) + assert.False(t, got.applied, "a claim in flight is left alone") ph := readSettlement(ctx, t, pool, id) - assert.Equal(t, ListingStatePending, ph.state, "the claim is left alone") + assert.Equal(t, ListingStatePending, ph.state) assert.Nil(t, ph.proceeds) + kept := readListingSettlement(ctx, t, pool, "ril-mhc-a") + require.NotNil(t, kept, "the claim keeps the listing") + assert.Nil(t, kept.proceeds) + requireMidClaimLeftAlone(ctx, t, store, pool, id, "ril-mhc-a") + + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-mhc-b", ListingStateActive)) + require.Contains(t, pollListings(ctx, t, store, id), "ril-mhc-a", "the listing is polled once the claim resolves") + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mhc-a", 88, 12)) + require.NoError(t, err) + assert.True(t, applied) requireListingProceeds(ctx, t, pool, "ril-mhc-a", 88) requireTotal(ctx, t, store, id, 88, 12, 1, 0) } +// A list attempt released with its token kept (an ambiguous AWS outcome) puts +// the row back in listing A's own state; if A's first claim left no row in +// the settlement table (it predates migration 000153), the retry's claim keeps +// A, so a relist does not lose it. +func checkResumedClaimKeepsListing(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { + const id = "mls-resume" + seedCanceledSeenListing(ctx, t, store, id, "ril-mrs-a") + // The state ReleaseMarketplaceListingClaim(keepAttempt=true) leaves: A's + // own state with the attempt's token and price schedule kept. Written with + // triggers off, as by a claim and release that ran before migration 000153. + tx, err := pool.Begin(ctx) + require.NoError(t, err) + t.Cleanup(func() { _ = tx.Rollback(ctx) }) + _, err = tx.Exec(ctx, `SET LOCAL session_replication_role = replica`) + require.NoError(t, err) + _, err = tx.Exec(ctx, `UPDATE purchase_history SET listing_client_token = 'tok-mrs', + listing_price_schedule = '[{"term_months":6,"price":40}]'::jsonb WHERE purchase_id = $1`, id) + require.NoError(t, err) + require.NoError(t, tx.Commit(ctx)) + require.Nil(t, readListingSettlement(ctx, t, pool, "ril-mrs-a"), "fixture: no row kept yet") + + claim, err := store.ClaimMarketplaceListingSlot(ctx, id, "tok-ignored", []byte(`[{"term_months":6,"price":40}]`)) + require.NoError(t, err) + require.NotNil(t, claim) + require.True(t, claim.Resumed) + kept := readListingSettlement(ctx, t, pool, "ril-mrs-a") + require.NotNil(t, kept, "the resumed claim keeps the listing") + assert.Equal(t, ListingStateCancelled, kept.status) + requireMidClaimLeftAlone(ctx, t, store, pool, id, "ril-mrs-a") + + require.NoError(t, store.UpdatePurchaseHistoryListing(ctx, id, "ril-mrs-b", ListingStateActive)) + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, "ril-mrs-a", 88, 12)) + require.NoError(t, err) + assert.True(t, applied) + requireTotal(ctx, t, store, id, 88, 12, 1, 0) +} + // Many pollers settling the same superseded listing at once record it exactly // once, with the winner's values. func checkConcurrentSupersededSettles(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { @@ -343,10 +408,12 @@ func checkConcurrentSupersededSettles(ctx context.Context, t *testing.T, store * // that listing's sale recorded exactly once, and no deadlock. In even rounds // all three start together; in odd rounds the polls start once the relist's // claim has committed (so the claim has kept the listing) and race the write -// of the new listing id. Both ways the listing's row can be created are hit: -// by a poll that settles it while the RI still holds it, and by the claim. +// of the new listing id. A poll that runs while the claim is in flight writes +// nothing (#1508); the listing must then still be due, and the next poll +// records it. Both ways the listing's row can be created are hit: by a poll +// that settles it while the RI still holds it, and by the claim. func checkConcurrentRelistAndPoll(ctx context.Context, t *testing.T, store *PostgresStore, pool *pgxpool.Pool) { - keptByClaim := 0 + keptByClaim, deferred := 0, 0 for round := range 20 { id := fmt.Sprintf("mls-race-%d", round) listingA := fmt.Sprintf("ril-mrr-a-%d", round) @@ -381,7 +448,18 @@ func checkConcurrentRelistAndPoll(ctx context.Context, t *testing.T, store *Post require.NoError(t, relistErr, "round %d", round) require.NoError(t, results[0].err, "round %d", round) require.NoError(t, results[1].err, "round %d", round) - assert.True(t, results[0].applied != results[1].applied, "round %d: exactly one poll records the sale", round) + require.False(t, results[0].applied && results[1].applied, "round %d: the sale is recorded at most once", round) + + // The next poll run: due exactly when no poll recorded the sale. + _, due := pollListings(ctx, t, store, id)[listingA] + recorded := results[0].applied || results[1].applied + assert.Equal(t, !recorded, due, "round %d: an unrecorded sale is still due", round) + applied, err := store.SettleMarketplaceListing(ctx, canceledSettlement(id, listingA, 88, 12)) + require.NoError(t, err) + assert.Equal(t, !recorded, applied, "round %d: the next poll records it only if no earlier poll did", round) + if !recorded { + deferred++ + } row := readListingSettlement(ctx, t, pool, listingA) require.NotNil(t, row, "round %d", round) @@ -395,7 +473,7 @@ func checkConcurrentRelistAndPoll(ctx context.Context, t *testing.T, store *Post keptByClaim++ } } - t.Logf("listing row created by the relist claim in %d of 20 rounds", keptByClaim) + t.Logf("listing row created by the relist claim in %d of 20 rounds; sale deferred past the claim in %d", keptByClaim, deferred) assert.GreaterOrEqual(t, keptByClaim, 10, "every odd round settles a listing the claim kept") } diff --git a/internal/config/store_postgres_marketplace_settlement.go b/internal/config/store_postgres_marketplace_settlement.go index 2555e2fe9..dbddebf6e 100644 --- a/internal/config/store_postgres_marketplace_settlement.go +++ b/internal/config/store_postgres_marketplace_settlement.go @@ -28,9 +28,12 @@ import ( // The first branch is the RI's current listing on purchase_history. The // second is a listing the RI has since moved on from (marketplace_listing_settlements, // issue #1494), selected by the same rules; its account, region and cloud -// account come from the purchase_history row it was made from. When both -// branches name a listing the purchase_history one is returned (src 0), so the -// poller sees the row's live state. +// account come from the purchase_history row it was made from. A listing a +// purchase_history row holds as a claim in flight (pending with a token) is +// left out of both branches until the claim resolves or is released, as +// #1508 leaves such a row alone. When both branches name a listing the +// purchase_history one is returned (src 0), so the poller sees the row's live +// state. const marketplaceListingsToPollSQL = ` SELECT DISTINCT ON (purchase_id, listing_id) account_id, purchase_id, provider, region, cloud_account_id::text, @@ -56,6 +59,11 @@ const marketplaceListingsToPollSQL = ` AND s.proceeds_received IS NULL AND (lower(s.status) IN ($1, $4) OR (lower(s.status) IN ($2, $3) AND s.listed_at IS NOT NULL)) + AND NOT EXISTS ( + SELECT 1 FROM purchase_history c + WHERE c.purchase_id = s.purchase_id AND c.listing_id = s.listing_id + AND lower(COALESCE(c.listing_state, '')) = $4 + AND c.listing_client_token IS NOT NULL) ) due ORDER BY purchase_id, listing_id, src, created_at, id ` @@ -68,7 +76,8 @@ const marketplaceListingsToPollSQL = ` // on purchase_history and the listings the RI was relisted from, which // marketplace_listing_settlements keeps (issue #1494). A pending row that // holds a list attempt's token is a claim in flight, whose listing_id is the -// RI's previous listing; that listing is polled from the settlement table. The +// RI's previous listing; that listing is not polled until the claim resolves +// (it is then polled from the settlement table) or is released. The // terminal 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 @@ -179,6 +188,19 @@ const upsertListingSettlementSQL = ` AND lower(s.status) IN ($7, $8, $3) ` +// listingHeldByClaimSQL share-locks the purchase_history rows that hold the +// listing and reports whether any of them is a list claim in flight (pending +// with a token). The lock makes a claim that starts now wait for this +// transaction, so no settlement is written while a row is mid-claim. +const listingHeldByClaimSQL = ` + SELECT COALESCE(bool_or(lower(COALESCE(listing_state, '')) = $3 + AND listing_client_token IS NOT NULL), false) + FROM (SELECT listing_state, listing_client_token + FROM purchase_history + WHERE purchase_id = $1 AND listing_id = $2 + FOR SHARE) held +` + // settleSupersededListingSQL records the settlement of a listing the RI no // longer holds on purchase_history, in the row the relist kept for it, under // the same guards as upsertListingSettlementSQL. @@ -202,9 +224,11 @@ const settleSupersededListingSQL = ` // marketplace_listing_settlements row (issue #1494), which is what per-RI // totals read. Each write applies only while that row has no proceeds // recorded and is active, pending with no list attempt in flight, 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 -// listing the RI was relisted from is still settled, in its own row. The +// in the same terminal state, and never while a purchase_history row holds the +// listing as a list claim in flight. So proceeds are recorded once however +// many polls race, a recorded cancel is never turned into a sale (or the +// reverse), a mid-claim row is left alone (#1508), and a listing the RI was +// relisted from is still settled, in its own row. The // purchase_history row is written first, so a poll and a relist of the same // RI take their locks in the same order. It reports whether either row was // written. @@ -249,13 +273,23 @@ func settleListingArgs(st MarketplaceListingSettlement) ([]any, error) { // settleMarketplaceListingTx runs SettleMarketplaceListing's writes in tx. A // listing the purchase_history row still holds gets its settlement row // created or filled; one the row no longer holds can only be filled in the -// settlement row its relist kept. +// settlement row its relist kept. Neither is written while a purchase_history +// row holds the listing as a claim in flight (#1508): the listing is settled +// once the claim resolves or is released. purchase_history is locked before +// the settlement table, the order a relist and its trigger take them in. func settleMarketplaceListingTx(ctx context.Context, tx pgx.Tx, args []any) (bool, error) { tag, err := tx.Exec(ctx, settleCurrentListingSQL, args...) if err != nil { return false, err } rowApplied := tag.RowsAffected() > 0 + var midClaim bool + if err := tx.QueryRow(ctx, listingHeldByClaimSQL, args[0], args[1], ListingStatePending).Scan(&midClaim); err != nil { + return false, err + } + if midClaim { + return rowApplied, nil + } listingSQL := settleSupersededListingSQL if rowApplied { listingSQL = upsertListingSettlementSQL diff --git a/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql b/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql index d47d662b3..b2310cef2 100644 --- a/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql +++ b/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql @@ -37,11 +37,15 @@ -- * Trigger purchase_history_keep_superseded_listing copies a listing into -- this table, as purchase_history last recorded it, at the moment the row -- stops describing it: when listing_id changes, or when a list attempt's --- claim sets listing_client_token (the claimed row keeps the previous --- listing id but its state becomes the attempt's). ON CONFLICT DO NOTHING +-- claim makes the row pending with a token (the claimed row keeps the +-- previous listing id but its state becomes the attempt's). A row that is +-- already pending with a token describes the attempt, not its listing id, +-- so it is never copied; a row whose attempt was released with its token +-- kept is back in the listing's own state and is. ON CONFLICT DO NOTHING -- keeps a row the poller already wrote. The poller keeps polling such a -- listing from here until it is settled, with the same rules it uses for --- purchase_history rows. +-- purchase_history rows, and like them leaves it alone while a row holds +-- it as a claim in flight (issue #1508). -- -- purchase_history's settlement columns keep their meaning: the latest -- listing's settlement, which the poller still writes in the same transaction. @@ -122,11 +126,12 @@ $$ LANGUAGE plpgsql; DROP TRIGGER IF EXISTS purchase_history_keep_superseded_listing ON purchase_history; CREATE TRIGGER purchase_history_keep_superseded_listing - AFTER UPDATE OF listing_id, listing_client_token ON purchase_history + AFTER UPDATE OF listing_id, listing_state, listing_client_token ON purchase_history FOR EACH ROW WHEN (COALESCE(OLD.listing_id, '') <> '' AND COALESCE(OLD.listing_state, '') <> '' - AND OLD.listing_client_token IS NULL + AND NOT (lower(OLD.listing_state) = 'pending' AND OLD.listing_client_token IS NOT NULL) AND (NEW.listing_id IS DISTINCT FROM OLD.listing_id - OR NEW.listing_client_token IS NOT NULL)) + OR (lower(COALESCE(NEW.listing_state, '')) = 'pending' + AND NEW.listing_client_token IS NOT NULL))) EXECUTE FUNCTION keep_superseded_marketplace_listing(); From 568c6d619da0f81e1d13c9fb5bf2e15f0d9891bd Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Wed, 28 Oct 2026 21:58:31 +0000 Subject: [PATCH 3/3] refactor(marketplace): read the mid-claim check through a helper Keeps govet's shadow check and gocritic's sloppyReassign both satisfied. Refs #1494 Co-Authored-By: Claude Opus 5.5 --- .../config/store_postgres_marketplace_settlement.go | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/internal/config/store_postgres_marketplace_settlement.go b/internal/config/store_postgres_marketplace_settlement.go index dbddebf6e..04816ae59 100644 --- a/internal/config/store_postgres_marketplace_settlement.go +++ b/internal/config/store_postgres_marketplace_settlement.go @@ -283,8 +283,8 @@ func settleMarketplaceListingTx(ctx context.Context, tx pgx.Tx, args []any) (boo return false, err } rowApplied := tag.RowsAffected() > 0 - var midClaim bool - if err := tx.QueryRow(ctx, listingHeldByClaimSQL, args[0], args[1], ListingStatePending).Scan(&midClaim); err != nil { + midClaim, err := listingHeldByClaim(ctx, tx, args[0], args[1]) + if err != nil { return false, err } if midClaim { @@ -301,6 +301,14 @@ func settleMarketplaceListingTx(ctx context.Context, tx pgx.Tx, args []any) (boo return rowApplied || tag.RowsAffected() > 0, nil } +// listingHeldByClaim runs listingHeldByClaimSQL in tx for the purchase and +// listing ids. +func listingHeldByClaim(ctx context.Context, tx pgx.Tx, purchaseID, listingID any) (bool, error) { + var midClaim bool + err := tx.QueryRow(ctx, listingHeldByClaimSQL, purchaseID, listingID, ListingStatePending).Scan(&midClaim) + return midClaim, err +} + // GetMarketplaceListingProceedsTotals returns, for each of purchaseIDs that // has any marketplace listing recorded, the net proceeds and AWS fees summed // over all of its settled listings, from marketplace_listing_settlements only