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

@@ -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
+}

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)
+	}
+}

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(), `
+	// Bounded try-lock instead of a blocking pg_advisory_lock: the SAME
-		SELECT pg_advisory_lock(hashtext('crussell:payment:' || $1))
+	// "crussell:payment:" key is held by the payment handlers across their full
-	`, bookingID); err != nil {
+	// 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))

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:<booking>" 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)
+	}
+}