popertots
e02567564e
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.
87 lines
2.6 KiB
Go
87 lines
2.6 KiB
Go
//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)
|
||
}
|
||
}
|