diff --git a/internal/api/handler_marketplace_poller.go b/internal/api/handler_marketplace_poller.go index 4ae65dec..d171081e 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 8105a164..f34129c3 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 af0fce80..cd617adf 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 895fa53e..798e9256 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 00000000..4604c359 --- /dev/null +++ b/internal/config/store_postgres_marketplace_listing_settlements_integration_test.go @@ -0,0 +1,487 @@ +//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("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) }) +} + +// 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") + 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) + 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) +} + +// 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 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") + 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.False(t, got.applied, "a claim in flight is left alone") + ph := readSettlement(ctx, t, pool, id) + 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) { + 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. 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, deferred := 0, 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) + 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) + 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; sale deferred past the claim in %d", keptByClaim, deferred) + 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 e9cbac6f..04816ae5 100644 --- a/internal/config/store_postgres_marketplace_settlement.go +++ b/internal/config/store_postgres_marketplace_settlement.go @@ -1,37 +1,46 @@ 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. 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, + 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 +50,42 @@ 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)) + 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 +` + +// 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 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 +// 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 +143,211 @@ 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) +` + +// 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. +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, 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. 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. 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 + midClaim, err := listingHeldByClaim(ctx, tx, args[0], args[1]) + if err != nil { + return false, err + } + if midClaim { + return rowApplied, nil + } + 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 +} + +// 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 +// (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 5772656e..44a8cb09 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 00000000..d1291c2c --- /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 00000000..b2310cef --- /dev/null +++ b/internal/database/postgres/migrations/000153_marketplace_listing_settlements.up.sql @@ -0,0 +1,137 @@ +-- 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 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, 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. +-- 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_state, listing_client_token ON purchase_history + FOR EACH ROW + WHEN (COALESCE(OLD.listing_id, '') <> '' + AND COALESCE(OLD.listing_state, '') <> '' + 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 (lower(COALESCE(NEW.listing_state, '')) = 'pending' + AND 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 00000000..27ed5489 --- /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 234510ef..ce62f622 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