From e02567564e2c956e42582e26bb41c813e93a1b7b Mon Sep 17 00:00:00 2001 From: Stephen Adamson Date: Mon, 3 Aug 2026 14:54:07 +0100 Subject: [PATCH] Add bounded advisory-lock helper; convert loyalty redemption to try-lock Introduce acquireAdvisoryLock (pg_try_advisory_lock with a ~3s bounded retry) and acquireAdvisoryXactLockBlocking (deliberately blocking for the cancellation-refund path where silently dropping a refund is worse than waiting). Convert ApplyLoyaltyRedemption to the bounded variant: concurrent redemptions during an in-flight payment return 409 instead of pinning a pool connection. Add uncontended + contended-timeout unit tests and a lock-contended 409 redemption test. --- backend/handlers/payments/locks.go | 96 +++++++++++++++++++++++ backend/handlers/payments/locks_test.go | 86 ++++++++++++++++++++ backend/handlers/payments/loyalty.go | 15 +++- backend/handlers/payments/loyalty_test.go | 60 ++++++++++++++ 4 files changed, 254 insertions(+), 3 deletions(-) create mode 100644 backend/handlers/payments/locks.go create mode 100644 backend/handlers/payments/locks_test.go diff --git a/backend/handlers/payments/locks.go b/backend/handlers/payments/locks.go new file mode 100644 index 0000000..de3128e --- /dev/null +++ b/backend/handlers/payments/locks.go @@ -0,0 +1,96 @@ +package payments + +import ( + "context" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +// advisoryLockAttempts and advisoryLockRetryDelay bound the total time a +// try-lock retry loop waits for a contended advisory lock (~3s). This is the +// core defence against pool exhaustion: every payment handler pins a pool +// connection and would otherwise block on `pg_advisory_lock` for the FULL +// Square round-trip (up to ~30s) of whichever request holds the lock, so a +// handful of concurrent same-key requests can exhaust the whole pool +// (max(4, numCPU)) and hang the app. With a bounded try-lock loop the waiter +// gives up after ~3s and surfaces "operation in progress" instead of holding a +// pool connection hostage. +const ( + advisoryLockAttempts = 30 + advisoryLockRetryDelay = 100 * time.Millisecond +) + +// queryRower is satisfied by both *pgxpool.Conn (session-level locks) and +// pgx.Tx (transaction-scoped locks) so the try-lock helpers work on pinned +// pool connections and inside transactions alike. +type queryRower interface { + QueryRow(ctx context.Context, sql string, args ...any) pgx.Row +} + +// acquireAdvisoryLock acquires a session advisory lock on the pinned conn with +// bounded retries (defends the pool against concurrent same-key lock waiters). +// Returns (true, nil) once the lock is held; (false, nil) if the lock could +// not be acquired within the bound — the caller must surface a 409/503 +// "operation in progress, try again" instead of blocking. The unlock is still +// the caller's responsibility (pg_advisory_unlock on the same conn via defer). +func acquireAdvisoryLock(ctx context.Context, conn *pgxpool.Conn, key string) (bool, error) { + return tryAdvisoryLock(ctx, conn, key, "pg_try_advisory_lock") +} + +// acquireAdvisoryXactLock is the transaction-scoped variant of +// acquireAdvisoryLock: the lock is held on the transaction and auto-released +// at commit/rollback, so the caller must NOT unlock explicitly. +func acquireAdvisoryXactLock(ctx context.Context, tx pgx.Tx, key string) (bool, error) { + return tryAdvisoryLock(ctx, tx, key, "pg_try_advisory_xact_lock") +} + +// acquireAdvisoryXactLockBlocking is the transaction-scoped BLOCKING variant +// of acquireAdvisoryXactLock: it issues `SELECT pg_advisory_xact_lock(...)` +// ONCE and waits for as long as the key is contended — there is no 3s bound. +// The lock is transaction-scoped, so it is auto-released at the caller's +// commit/rollback (never unlocked explicitly). +// +// This stays deliberately blocking for ONE site only: the admin cancellation +// path (lockCancellationPayments). Admin cancellations are rare and there is +// only ONE such transaction at a time, so the pool-exhaustion rationale that +// justifies the bounded try-lock everywhere else does not apply here. A bound +// would be actively harmful: if a manual RefundPayment holds the same +// "crussell:refund:" key across its up-to-30s Square round-trip, a timed-out +// cancellation would abort, and the caller (manage.go) would commit the +// cancellation with ZERO refund rows created — no sweep retry is possible +// because the rows never existed, so the refund would be silently lost. The +// blocking acquire guarantees the cancellation refund runs, waiting for the +// manual refund to finish rather than dropping the money. +func acquireAdvisoryXactLockBlocking(ctx context.Context, tx pgx.Tx, key string) error { + // Exec, not QueryRow.Scan: pg_advisory_xact_lock returns void, and Exec + // discards the result set (the same pattern the payment handlers use for + // their blocking pg_advisory_lock) while still blocking server-side until + // the lock is granted. + _, err := tx.Exec(ctx, `SELECT pg_advisory_xact_lock(hashtext($1))`, key) + return err +} + +// tryAdvisoryLock is the shared try-lock retry loop. fn is the Postgres +// advisory-lock function to call (pg_try_advisory_lock or +// pg_try_advisory_xact_lock). The key is passed to hashtext() so it is hashed +// to a bigint exactly like the blocking `pg_advisory_lock(hashtext($1))` calls +// it replaces — the two acquire the same locks. +func tryAdvisoryLock(ctx context.Context, q queryRower, key, fn string) (bool, error) { + for attempt := 0; attempt < advisoryLockAttempts; attempt++ { // ~3s total + var acquired bool + if err := q.QueryRow(ctx, `SELECT `+fn+`(hashtext($1))`, key).Scan(&acquired); err != nil { + return false, err + } + if acquired { + return true, nil + } + select { + case <-time.After(advisoryLockRetryDelay): + case <-ctx.Done(): + return false, ctx.Err() + } + } + return false, nil +} diff --git a/backend/handlers/payments/locks_test.go b/backend/handlers/payments/locks_test.go new file mode 100644 index 0000000..41d1322 --- /dev/null +++ b/backend/handlers/payments/locks_test.go @@ -0,0 +1,86 @@ +//go:build test && dev + +package payments + +import ( + "context" + "testing" + "time" + + "crussell/db" +) + +// TestAdvisoryLock_Uncontended_Acquires verifies the happy path: an unlocked +// key is acquired immediately with a bounded try-lock. +func TestAdvisoryLock_Uncontended_Acquires(t *testing.T) { + t.Parallel() + ctx := context.Background() + + conn, err := db.Conn.Acquire(ctx) + if err != nil { + t.Fatalf("failed to acquire pool connection: %v", err) + } + defer conn.Release() + + key := "crussell:payment:locktest-uncontended" + acquired, err := acquireAdvisoryLock(ctx, conn, key) + if err != nil { + t.Fatalf("acquireAdvisoryLock returned error: %v", err) + } + if !acquired { + t.Fatal("expected uncontended lock to be acquired") + } + if _, err := conn.Exec(ctx, `SELECT pg_advisory_unlock(hashtext($1))`, key); err != nil { + t.Fatalf("failed to release lock: %v", err) + } +} + +// TestAdvisoryLock_Contended_TimesOutWithinBound verifies the bounded try-lock +// timeout branch: when another connection holds the same key, acquireAdvisoryLock +// returns (false, nil) after the ~3s retry bound instead of blocking forever +// (the pool-exhaustion defence). +func TestAdvisoryLock_Contended_TimesOutWithinBound(t *testing.T) { + t.Parallel() + ctx := context.Background() + + holder, err := db.Conn.Acquire(ctx) + if err != nil { + t.Fatalf("failed to acquire holder connection: %v", err) + } + defer holder.Release() + + key := "crussell:payment:locktest-contended" + // Hold the advisory lock on a dedicated connection so every try-lock + // attempt from the second connection fails. + if _, err := holder.Exec(ctx, `SELECT pg_advisory_lock(hashtext($1))`, key); err != nil { + t.Fatalf("failed to acquire holder lock: %v", err) + } + defer func() { + _, _ = holder.Exec(context.Background(), `SELECT pg_advisory_unlock(hashtext($1))`, key) + }() + + waiter, err := db.Conn.Acquire(ctx) + if err != nil { + t.Fatalf("failed to acquire waiter connection: %v", err) + } + defer waiter.Release() + + start := time.Now() + acquired, err := acquireAdvisoryLock(ctx, waiter, key) + elapsed := time.Since(start) + + if err != nil { + t.Fatalf("acquireAdvisoryLock returned error (expected false, nil): %v", err) + } + if acquired { + t.Fatal("expected contended lock NOT to be acquired") + } + // The bound is 30 attempts × 100ms ≈ 3s. Assert it gave up within a sane + // window (did not hang) and did not return prematurely. + if elapsed < 2*time.Second { + t.Errorf("expected the timeout bound (~3s) to elapse before giving up, returned after %v", elapsed) + } + if elapsed > 10*time.Second { + t.Errorf("expected to give up within the ~3s bound, took %v", elapsed) + } +} diff --git a/backend/handlers/payments/loyalty.go b/backend/handlers/payments/loyalty.go index 5d547f2..699ddfd 100644 --- a/backend/handlers/payments/loyalty.go +++ b/backend/handlers/payments/loyalty.go @@ -95,13 +95,22 @@ func ApplyLoyaltyRedemption(w http.ResponseWriter, r *http.Request) { return } defer pinConn.Release() - if _, err := pinConn.Exec(r.Context(), ` - SELECT pg_advisory_lock(hashtext('crussell:payment:' || $1)) - `, bookingID); err != nil { + // Bounded try-lock instead of a blocking pg_advisory_lock: the SAME + // "crussell:payment:" key is held by the payment handlers across their full + // Square round-trip (~30s), so a blocking acquire here would pin this pool + // connection for that long — a handful of concurrent redemption requests + // during an in-flight payment would exhaust the pool (max(4, numCPU)) and + // hang the app. Give up after ~3s and surface 409 instead. + lockOK, err := acquireAdvisoryLock(r.Context(), pinConn, "crussell:payment:"+bookingID) + if err != nil { log.Printf("Failed to acquire loyalty redemption serialization lock for %s: %v", bookingID, err) http.Error(w, "internal server error", http.StatusInternalServerError) return } + if !lockOK { + http.Error(w, "Another payment operation is in progress, try again", http.StatusConflict) + return + } defer func() { if _, err := pinConn.Exec(context.Background(), ` SELECT pg_advisory_unlock(hashtext('crussell:payment:' || $1)) diff --git a/backend/handlers/payments/loyalty_test.go b/backend/handlers/payments/loyalty_test.go index 018747a..760d3b2 100644 --- a/backend/handlers/payments/loyalty_test.go +++ b/backend/handlers/payments/loyalty_test.go @@ -707,3 +707,63 @@ func TestApplyLoyaltyRedemption_NoPendingRedemption(t *testing.T) { t.Errorf("expected 400, got %d: %s", w.Code, w.Body.String()) } } + +// TestApplyLoyaltyRedemption_LockContended_Returns409 verifies the bounded +// try-lock defence (loyalty.go): while another connection holds the +// "crussell:payment:" advisory lock (e.g. an in-flight payment), a +// redemption attempt must NOT block the pool connection — it gives up after +// the ~3s bound and surfaces a 409 instead. +func TestApplyLoyaltyRedemption_LockContended_Returns409(t *testing.T) { + t.Parallel() + ctx, tx := testutils.SetupTestTx(t) + _, bookingID, userToken := setupLoyaltyUser(t, ctx, tx, 10) + + // Commit the setup tx so the handler sees committed rows. + innerTx := db.TxFromContext(ctx) + if innerTx == nil { + t.Fatal("no transaction in context") + } + if err := innerTx.Commit(ctx); err != nil { + t.Fatalf("failed to commit setup tx: %v", err) + } + + // Hold the booking-payment advisory lock on a dedicated pinned connection + // so every try-lock attempt from the handler's connection fails. + holder, err := db.Conn.Acquire(context.Background()) + if err != nil { + t.Fatalf("failed to acquire holder connection: %v", err) + } + defer holder.Release() + if _, err := holder.Exec(context.Background(), + `SELECT pg_advisory_lock(hashtext('crussell:payment:' || $1))`, bookingID); err != nil { + t.Fatalf("failed to acquire holder lock: %v", err) + } + defer func() { + _, _ = holder.Exec(context.Background(), + `SELECT pg_advisory_unlock(hashtext('crussell:payment:' || $1))`, bookingID) + }() + + start := time.Now() + w := makeApplyRedemptionRequest(bookingID, userToken, context.Background()) + elapsed := time.Since(start) + + if w.Code != http.StatusConflict { + t.Fatalf("expected 409 on contended lock, got %d: %s", w.Code, w.Body.String()) + } + // The bound is 30×100ms ≈ 3s. It must give up within a sane window (did + // not block forever on the pool) — allow generous CI headroom. + if elapsed > 15*time.Second { + t.Errorf("lock contention should give up after ~3s, took %v", elapsed) + } + + // Redemption must NOT have been applied. + var discountCount int + err = db.Conn.QueryRow(context.Background(), + "SELECT COUNT(*) FROM booking_discounts WHERE booking_id = $1 AND discount_source = 'loyalty'", bookingID).Scan(&discountCount) + if err != nil { + t.Fatalf("failed to query booking_discounts: %v", err) + } + if discountCount != 0 { + t.Errorf("expected 0 discounts (redemption rejected), got %d", discountCount) + } +}