From 2d22986395172c2720c86d7623e2de4bc3420b1f Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 29 Oct 2026 01:12:10 +0000 Subject: [PATCH 1/5] test: gate the Archera shared-build test with synctest instead of a 100ms sleep TestClient_ConcurrentCallersShareOneBuild relied on a resolver that slept 100ms so the 20 racing callers would overlap the build. It now parks the resolver on a gate channel inside a synctest bubble: synctest.Wait proves every caller is blocked (one in the resolver, the rest on the build slot) before the test checks that only one build started, then opens the gate and checks the waiters reused the cached client. Removing the semaphore or the re-check after acquire fails it 50 out of 50 runs. The frontend index test waited 10ms for an init call the DOMContentLoaded listener makes synchronously; it now asserts directly. The no-wallclock- sleeps guard missed that form (assertions inside a setTimeout block that then resolves), so it now flags it too. Closes #36 Co-Authored-By: Claude Opus 5.5 --- frontend/src/__tests__/index.test.ts | 7 +- .../src/__tests__/no-wallclock-sleeps.test.ts | 8 ++- internal/archera/provider_cancel_test.go | 69 +++++++++++-------- 3 files changed, 51 insertions(+), 33 deletions(-) diff --git a/frontend/src/__tests__/index.test.ts b/frontend/src/__tests__/index.test.ts index 202081a99..1e21073d3 100644 --- a/frontend/src/__tests__/index.test.ts +++ b/frontend/src/__tests__/index.test.ts @@ -117,11 +117,8 @@ describe('Index Module', () => { const event = new Event('DOMContentLoaded'); document.dispatchEvent(event); - // Wait for async init - return new Promise(resolve => setTimeout(() => { - expect(app.init).toHaveBeenCalled(); - resolve(undefined); - }, 10)); + // The listener calls init synchronously; no wait is needed (issue #36). + expect(app.init).toHaveBeenCalled(); }); test('wires the wide-table scroll affordances on DOMContentLoaded (#185)', () => { diff --git a/frontend/src/__tests__/no-wallclock-sleeps.test.ts b/frontend/src/__tests__/no-wallclock-sleeps.test.ts index 253ca67fb..3b8d3bd66 100644 --- a/frontend/src/__tests__/no-wallclock-sleeps.test.ts +++ b/frontend/src/__tests__/no-wallclock-sleeps.test.ts @@ -17,6 +17,9 @@ const testsDir = __dirname; // `new Promise(r => setTimeout(r, N))` and `new Promise(r => setTimeout(() => r(), N))`, // with or without a type argument or parentheses around the parameter. const SLEEP = /new Promise\s*(?:<[^>]*>)?\(\s*\(?\s*([A-Za-z_$][\w$]*)\s*\)?\s*=>\s*\{?\s*setTimeout\(\s*(?:\1|\(\)\s*=>\s*\1\(\))\s*,\s*([^)]+?)\s*\)/g; +// `new Promise(r => setTimeout(() => { expect(...); r(); }, N))`: the +// assertions run inside the timer body, after the same guessed wait. +const SLEEP_BLOCK = /new Promise\s*(?:<[^>]*>)?\(\s*\(?\s*([A-Za-z_$][\w$]*)\s*\)?\s*=>\s*\{?\s*setTimeout\(\s*\(\)\s*=>\s*\{(?:(?!setTimeout\()[\s\S])*?\b\1\([^)]*\);?\s*\}\s*,\s*([^)]+?)\s*\)/g; function testSources(dir: string): string[] { return fs.readdirSync(dir, { withFileTypes: true }).flatMap((entry) => { @@ -28,7 +31,7 @@ function testSources(dir: string): string[] { function findSleeps(source: string): { line: number; delay: string }[] { const found: { line: number; delay: string }[] = []; - for (const match of source.matchAll(SLEEP)) { + for (const match of [...source.matchAll(SLEEP), ...source.matchAll(SLEEP_BLOCK)]) { const delay = match[2] ?? ''; if (/^0+$/.test(delay)) continue; found.push({ line: source.slice(0, match.index).split('\n').length, delay }); @@ -46,6 +49,9 @@ describe('no wall-clock sleeps in unit tests (issue #36)', () => { expect(findSleeps('\nawait new Promise(r => { setTimeout(r, 5); });')).toEqual([{ line: 2, delay: '5' }]); expect(findSleeps('await new Promise((resolve) => setTimeout(resolve, 0));')).toEqual([]); expect(findSleeps('setTimeout(() => closeModal(modal), 2000);')).toEqual([]); + expect(findSleeps('return new Promise(resolve => setTimeout(() => {\n expect(x).toBe(1);\n resolve(undefined);\n}, 10));')) + .toEqual([{ line: 1, delay: '10' }]); + expect(findSleeps('await new Promise((done) => setTimeout(() => { check(); done(); }, 0));')).toEqual([]); }); test('no test file sleeps for a fixed time to wait for async work', () => { diff --git a/internal/archera/provider_cancel_test.go b/internal/archera/provider_cancel_test.go index 4cefc693a..1ed9d854a 100644 --- a/internal/archera/provider_cancel_test.go +++ b/internal/archera/provider_cancel_test.go @@ -7,6 +7,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/logging" @@ -100,40 +101,54 @@ func TestClient_CancelledCallerWinningTheSlotNeverResolves(t *testing.T) { } } -// slowResolver counts calls and sleeps so racing callers overlap the build. -type slowResolver struct { +// gatedResolver counts calls and parks each one until release is closed, so a +// test decides when the in-flight build may finish. +type gatedResolver struct { flakyResolver - n atomic.Int32 + n atomic.Int32 + release chan struct{} } -func (s *slowResolver) GetSecret(context.Context, string) (string, error) { - s.n.Add(1) - time.Sleep(100 * time.Millisecond) +func (g *gatedResolver) GetSecret(context.Context, string) (string, error) { + g.n.Add(1) + <-g.release return testKey, nil } +// The synctest bubble makes the overlap explicit instead of hoping a sleep is +// long enough: synctest.Wait returns only once every caller is durably +// blocked, i.e. one is parked inside the resolver and the rest on the build +// slot. A second resolver call at that point means the build is not +// serialized; one after release means waiters rebuild instead of reusing the +// cached client. func TestClient_ConcurrentCallersShareOneBuild(t *testing.T) { - r := &slowResolver{} - p := NewProvider(fullSettings(), r, nil) - const n = 20 - got := make([]any, n) - var wg sync.WaitGroup - for i := 0; i < n; i++ { - wg.Add(1) - go func() { - defer wg.Done() - c, err := p.Client(context.Background()) - if err == nil { - got[i] = c - } - }() - } - wg.Wait() - require.EqualValues(t, 1, r.n.Load(), "resolver called exactly once") - require.NotNil(t, got[0]) - for i := range got { - require.Same(t, got[0], got[i]) - } + synctest.Test(t, func(t *testing.T) { + r := &gatedResolver{release: make(chan struct{})} + p := NewProvider(fullSettings(), r, nil) + const n = 20 + got := make([]any, n) + errs := make([]error, n) + var wg sync.WaitGroup + openGate := sync.OnceFunc(func() { close(r.release) }) + // A failed assertion exits this goroutine; drain the callers so the + // bubble reports the failure rather than a deadlock. + defer func() { openGate(); wg.Wait() }() + for i := 0; i < n; i++ { + wg.Go(func() { got[i], errs[i] = p.Client(context.Background()) }) + } + synctest.Wait() + require.EqualValues(t, 1, r.n.Load(), "only one caller may build while the others wait") + require.Nil(t, p.client.Load(), "nothing cached before the build completes") + + openGate() + wg.Wait() + require.EqualValues(t, 1, r.n.Load(), "waiters reuse the built client instead of rebuilding") + for i := range got { + require.NoError(t, errs[i]) + require.NotNil(t, got[i]) + require.Same(t, got[0], got[i]) + } + }) } // ctxBlockedResolver parks until the caller's ctx ends, then returns the From 60a8c8d88bfccc37c81ffb1a627569fc41dfd068 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 29 Oct 2026 03:29:13 +0000 Subject: [PATCH 2/5] test(config): observe lock contention instead of a quiet window in three race tests The claim, direct-finalize and purchase-history race tests proved "the contender waits on the uncommitted holder" by checking that it had not returned after 300-500ms. A contender that never contends but is slow to run (a loaded runner) passed that window and then answered after the holder committed. Verified: a claim that sleeps 400ms and answers from a read passes the old test and fails the new one; the same holds for the finalize and the history save. testhelpers.AwaitBlockedBy now polls pg_locks until a backend waits on a lock the holder's session holds (pg_blocking_pids), and fails at once if the contender returns first. testhelpers.AwaitArrivalsOrLockWaits and LockWaiters back the race barriers converted next. Refs #36 Co-Authored-By: Claude Opus 5.5 --- internal/config/store_postgres_db_test.go | 11 ++- ...ostgres_history_dedupe_integration_test.go | 7 +- ...ore_postgres_ri_exchange_direct_db_test.go | 6 +- .../database/postgres/testhelpers/lockwait.go | 84 +++++++++++++++++++ 4 files changed, 96 insertions(+), 12 deletions(-) diff --git a/internal/config/store_postgres_db_test.go b/internal/config/store_postgres_db_test.go index 72c66efa5..36f906411 100644 --- a/internal/config/store_postgres_db_test.go +++ b/internal/config/store_postgres_db_test.go @@ -1476,6 +1476,7 @@ func TestPostgresStoreDB_ClaimRIExchangeIdempotencyKey(t *testing.T) { _, err = tx.Exec(ctx, `INSERT INTO ri_exchange_idempotency (idempotency_key, claimed_at) VALUES ($1, now())`, key) require.NoError(t, err) + holderPID := testhelpers.BackendPID(ctx, t, tx) type claimResult struct { claimed bool @@ -1488,15 +1489,13 @@ func TestPostgresStoreDB_ClaimRIExchangeIdempotencyKey(t *testing.T) { }() // The claim must contend for the row rather than answer from a read, - // so it cannot return while the conflicting write is uncommitted. The - // wait is the only way to observe "still blocked"; it is a lower bound - // on the observation, not a sleep the assertion depends on. - select { - case r := <-done: + // so it cannot return while the conflicting write is uncommitted. + // Observing it queued behind tx proves that; a quiet window could not + // tell a contender that is blocked from one that has not run yet. + if r, answered := testhelpers.AwaitBlockedBy(ctx, t, conn.Pool(), holderPID, done); answered { _ = tx.Rollback(ctx) t.Fatalf("the claim answered %v while a conflicting claim was still uncommitted; "+ "it read the ledger instead of contending for the row", r.claimed) - case <-time.After(300 * time.Millisecond): } require.NoError(t, tx.Commit(ctx)) diff --git a/internal/config/store_postgres_history_dedupe_integration_test.go b/internal/config/store_postgres_history_dedupe_integration_test.go index c89e1f071..6bfd0d083 100644 --- a/internal/config/store_postgres_history_dedupe_integration_test.go +++ b/internal/config/store_postgres_history_dedupe_integration_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -101,14 +102,14 @@ func TestSavePurchaseHistory_ConcurrentSaveWaitsForUncommittedTwin(t *testing.T) VALUES ('123456789012', 'ri-race', now(), 'aws', 'ec2', 'us-east-1', 'm5.large', 1, 'all-upfront')`) require.NoError(t, err) + holderPID := testhelpers.BackendPID(ctx, t, tx) + done := make(chan error, 1) go func() { done <- store.SavePurchaseHistory(ctx, historyFixture("123456789012", "ri-race")) }() - select { - case err := <-done: + if err, answered := testhelpers.AwaitBlockedBy(ctx, t, conn.Pool(), holderPID, done); answered { _ = tx.Rollback(ctx) t.Fatalf("save returned (%v) while a twin was uncommitted; it must wait on the unique index", err) - case <-time.After(500 * time.Millisecond): } require.NoError(t, tx.Commit(ctx)) select { diff --git a/internal/config/store_postgres_ri_exchange_direct_db_test.go b/internal/config/store_postgres_ri_exchange_direct_db_test.go index a3455f1b8..19afe3d9c 100644 --- a/internal/config/store_postgres_ri_exchange_direct_db_test.go +++ b/internal/config/store_postgres_ri_exchange_direct_db_test.go @@ -9,6 +9,7 @@ import ( "testing" "time" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" "github.com/google/uuid" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" @@ -148,16 +149,15 @@ func TestPostgresStoreDB_RIExchangeDirectExecution(t *testing.T) { WHERE id = $1 AND status = 'submitted'`, rec.ID) require.NoError(t, err) require.EqualValues(t, 1, tag.RowsAffected()) + holderPID := testhelpers.BackendPID(ctx, t, tx) done := make(chan error, 1) go func() { done <- store.FinalizeRIExchangeDirectExecution(ctx, rec.ID, RIExchangeDirectOutcome{Status: RIExchangeDirectStatusFailed, Error: "second"}) }() - select { - case err := <-done: + if err, answered := testhelpers.AwaitBlockedBy(ctx, t, conn.Pool(), holderPID, done); answered { _ = tx.Rollback(ctx) t.Fatalf("the second finalize answered (%v) while the first was uncommitted; it must contend for the row", err) - case <-time.After(300 * time.Millisecond): } require.NoError(t, tx.Commit(ctx)) diff --git a/internal/database/postgres/testhelpers/lockwait.go b/internal/database/postgres/testhelpers/lockwait.go index 8917f5741..0d956e9ca 100644 --- a/internal/database/postgres/testhelpers/lockwait.go +++ b/internal/database/postgres/testhelpers/lockwait.go @@ -3,6 +3,7 @@ package testhelpers import ( "context" "errors" + "fmt" "testing" "time" @@ -102,3 +103,86 @@ func Await[T any](t *testing.T, done <-chan T) T { panic("unreachable") } } + +// lockPollInterval and lockPollTimeout bound the lock observations below. The +// timeout only fails a stuck test; a passing run returns as soon as the +// observed state appears. +const ( + lockPollInterval = 5 * time.Millisecond + lockPollTimeout = 30 * time.Second +) + +// BackendPID returns the server process ID of tx's session, the holderPID that +// AwaitBlockedBy takes. +func BackendPID(ctx context.Context, t *testing.T, tx pgx.Tx) int { + t.Helper() + var pid int + require.NoError(t, tx.QueryRow(ctx, `SELECT pg_backend_pid()`).Scan(&pid)) + return pid +} + +// AwaitBlockedBy proves a contender is queued behind holderPID's transaction: +// it returns once some backend waits on a lock that holderPID holds, so the +// contender cannot finish before the holder ends. If done delivers first, the +// contender answered without contending; AwaitBlockedBy returns that value and +// true. A fixed quiet window (select on done against time.After) cannot tell +// "blocked" from "not there yet", so on a slow runner it passes a contender +// that never contends. This observation cannot. +// +// Query db from a connection other than the holder's: pg_locks is read live, +// but the holder's session is busy holding the lock. +func AwaitBlockedBy[T any](ctx context.Context, t *testing.T, db LockDB, holderPID int, done <-chan T) (T, bool) { + t.Helper() + deadline := time.Now().Add(lockPollTimeout) + for time.Now().Before(deadline) { + var blocked bool + require.NoError(t, db.QueryRow(ctx, `SELECT EXISTS ( + SELECT 1 FROM pg_locks WHERE NOT granted AND $1::int = ANY(pg_blocking_pids(pid)))`, holderPID).Scan(&blocked)) + if blocked { + var zero T + return zero, false + } + select { + case v := <-done: + return v, true + case <-time.After(lockPollInterval): + } + } + t.Fatalf("no backend blocked on backend %d within %s and the contender did not return", holderPID, lockPollTimeout) + panic("unreachable") +} + +// LockWaiters counts backends of this database that wait on a lock in a +// statement matching queryLike (a SQL LIKE pattern). +func LockWaiters(ctx context.Context, db LockDB, queryLike string) (int, error) { + var n int + err := db.QueryRow(ctx, `SELECT count(*) FROM pg_stat_activity WHERE datname = current_database() + AND wait_event_type = 'Lock' AND query LIKE $1`, queryLike).Scan(&n) + return n, err +} + +// AwaitArrivalsOrLockWaits is a race-test barrier: it returns once each of +// parties has either arrived (arrived reports how many have) or is waiting on +// a lock in a statement matching queryLike. Under a serializing lock the +// parties that have not arrived are provably parked on it, so waiting longer +// cannot change the outcome; without the lock every party arrives. Either way +// the barrier opens on observed state, never on a guessed timeout. It returns +// an error rather than failing a test so the goroutine under test can call it. +func AwaitArrivalsOrLockWaits(ctx context.Context, db LockDB, parties int, arrived func() int, queryLike string) error { + deadline := time.Now().Add(lockPollTimeout) + for time.Now().Before(deadline) { + // Read arrivals first: a party that arrives between the two reads is + // then missed for one round instead of counted twice. + n := arrived() + waiting, err := LockWaiters(ctx, db, queryLike) + if err != nil { + return err + } + if n+waiting >= parties { + return nil + } + time.Sleep(lockPollInterval) + } + return fmt.Errorf("barrier: %d of %d parties neither arrived nor waited on a %q lock within %s", + parties-arrived(), parties, queryLike, lockPollTimeout) +} From bd3e973d70b58cfcd22f609cd1ff11b826b852e1 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 29 Oct 2026 03:34:46 +0000 Subject: [PATCH 3/5] test(api): observe lock waits instead of fixed windows in the ladder and submit race tests TestRetry_TrancheLockBlocksReconciler and TestLadderReconcile_TailLockBlocksRealRetryLink proved "the other side waits on the held row" by checking it had not returned after 1.5s. They now record the paused transaction's backend and wait until a backend is blocked by it (testhelpers.AwaitBlockedBy), failing at once if the other side returns first. Dropping FOR UPDATE from LockLadderTrancheForRetryTx or from the chain lock fails the matching test. insertBarrierStore opened its barrier after a fixed 300ms/1s wait, so on a slow runner an unguarded second submit could miss it and see the first one's committed row. It now opens once both submits arrived or the missing one is observed parked on the per-creator advisory lock. Dropping that lock fails both tests that use the barrier. Refs #36 Co-Authored-By: Claude Opus 5.5 --- ...ladder_reconcile_retry_integration_test.go | 17 ++++++++--- ...purchases_retry_ladder_integration_test.go | 11 +++++-- ...ases_retry_submit_race_integration_test.go | 7 +++-- ...urchases_submit_dedupe_integration_test.go | 29 ++++++++++--------- 4 files changed, 40 insertions(+), 24 deletions(-) diff --git a/internal/api/handler_ladder_reconcile_retry_integration_test.go b/internal/api/handler_ladder_reconcile_retry_integration_test.go index 881c2e72a..25955673e 100644 --- a/internal/api/handler_ladder_reconcile_retry_integration_test.go +++ b/internal/api/handler_ladder_reconcile_retry_integration_test.go @@ -12,9 +12,11 @@ import ( "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" "github.com/LeanerCloud/cloud-commitments-platform/internal/purchase" "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -32,10 +34,14 @@ type pausedChainStore struct { *config.PostgresStore locked chan struct{} release chan struct{} + pid int // backend holding the chain locks; set before locked closes } func (s *pausedChainStore) LockLadderTrancheChainTx(ctx context.Context, tx pgx.Tx, id string) ([]config.LadderChainNode, error) { chain, err := s.PostgresStore.LockLadderTrancheChainTx(ctx, tx, id) + if err == nil { + err = tx.QueryRow(ctx, `SELECT pg_backend_pid()`).Scan(&s.pid) + } close(s.locked) <-s.release return chain, err @@ -43,6 +49,7 @@ func (s *pausedChainStore) LockLadderTrancheChainTx(ctx context.Context, tx pgx. type ladderRetryRig struct { store *config.PostgresStore + pool *pgxpool.Pool cfgID string trancheID string failedExec *config.PurchaseExecution @@ -60,7 +67,7 @@ func newLadderRetryRig(t *testing.T) *ladderRetryRig { Mode: "email_approval", Cadence: "daily", TargetCoverage: 80, BufferFraction: 0.1, BaselinePercentile: 5, LookbackDays: 30, BufferUtilizationThreshold: 50, MaxActionsPerRun: 5, RampSchedule: []byte(`{"steps":[{"after_days":0,"fraction":1.0}]}`)}) require.NoError(t, err) - r := &ladderRetryRig{store: store, cfgID: cfg.ID} + r := &ladderRetryRig{store: store, pool: pool, cfgID: cfg.ID} r.addFiredTranche(t) return r } @@ -133,10 +140,12 @@ func TestLadderReconcile_TailLockBlocksRealRetryLink(t *testing.T) { retried := make(chan error, 1) go func() { retried <- h.persistRetrySuccessorTx(ctx, r.failedExec, r.successor(), nil) }() - select { - case err := <-retried: + // The retry must queue behind the reconciler's chain locks; observing it + // blocked proves it cannot link before the reconciler decides. + if err, finished := testhelpers.AwaitBlockedBy(ctx, t, r.pool, paused.pid, retried); finished { + close(paused.release) + <-reconciled t.Fatalf("retry linked while the reconciler held the tail row (err=%v)", err) - case <-time.After(1500 * time.Millisecond): } close(paused.release) diff --git a/internal/api/handler_purchases_retry_ladder_integration_test.go b/internal/api/handler_purchases_retry_ladder_integration_test.go index 96237c75a..11bec9742 100644 --- a/internal/api/handler_purchases_retry_ladder_integration_test.go +++ b/internal/api/handler_purchases_retry_ladder_integration_test.go @@ -15,6 +15,7 @@ import ( "github.com/LeanerCloud/cloud-commitments-go/pkg/ladder" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" "github.com/LeanerCloud/cloud-commitments-platform/internal/purchase" "github.com/google/uuid" "github.com/jackc/pgx/v5" @@ -102,10 +103,14 @@ type retryLockPausedStore struct { *config.PostgresStore locked chan struct{} release chan struct{} + pid int // backend holding the tranche row; set before locked closes } func (s *retryLockPausedStore) LockLadderTrancheForRetryTx(ctx context.Context, tx pgx.Tx, key string) (ladder.TrancheStatus, bool, error) { st, found, err := s.PostgresStore.LockLadderTrancheForRetryTx(ctx, tx, key) + if err == nil { + err = tx.QueryRow(ctx, `SELECT pg_backend_pid()`).Scan(&s.pid) + } close(s.locked) <-s.release return st, found, err @@ -129,12 +134,12 @@ func TestRetry_TrancheLockBlocksReconciler(t *testing.T) { _, err := purchase.ReconcileLadderTranches(ctx, r.store, visibleNever{}, time.Now()) reconciled <- err }() - select { - case err := <-reconciled: + // The reconciler must queue behind Retry's tranche row lock; observing it + // blocked proves it cannot fail the tranche before Retry links. + if err, finished := testhelpers.AwaitBlockedBy(ctx, t, r.pool, paused.pid, reconciled); finished { close(paused.release) // let the Retry goroutine finish so cleanup cannot hang <-retried t.Fatalf("reconciler finished while Retry held the tranche row (err=%v)", err) - case <-time.After(1500 * time.Millisecond): } close(paused.release) diff --git a/internal/api/handler_purchases_retry_submit_race_integration_test.go b/internal/api/handler_purchases_retry_submit_race_integration_test.go index 07711112d..e9fd7f453 100644 --- a/internal/api/handler_purchases_retry_submit_race_integration_test.go +++ b/internal/api/handler_purchases_retry_submit_race_integration_test.go @@ -91,8 +91,9 @@ func countPendingRows(t *testing.T, pg *testhelpers.PostgresContainer) int { // deadlock (40P01). Rounds alternate which side starts first. // // The forced case holds each transaction at its INSERT until both have -// arrived (or 300ms pass), so without the lock both pass their checks before -// either inserts; its 50ms head start makes each side win half the rounds. +// arrived or the other is parked on the per-creator lock, so without the lock +// both pass their checks before either inserts; its 50ms head start makes +// each side win half the rounds. // The free case races with a sub-millisecond random lag. func TestRetrySuccessorRacesFreshSubmit(t *testing.T) { pg, store := setupSubmitDB(t) @@ -111,7 +112,7 @@ func TestRetrySuccessorRacesFreshSubmit(t *testing.T) { lag func() time.Duration bothOrders bool }{ - {"forced interleaving", &Handler{config: &insertBarrierStore{PostgresStore: store, arrived: make(chan struct{}, 2), parties: 2, wait: 300 * time.Millisecond}}, + {"forced interleaving", &Handler{config: &insertBarrierStore{PostgresStore: store, db: pg.DB.Pool(), arrived: make(chan struct{}, 2), parties: 2}}, 8, func() time.Duration { return 50 * time.Millisecond }, true}, {"free race", &Handler{config: store}, 100, func() time.Duration { return time.Duration(rand.IntN(500)) * time.Microsecond }, false}, } { diff --git a/internal/api/handler_purchases_submit_dedupe_integration_test.go b/internal/api/handler_purchases_submit_dedupe_integration_test.go index 7d7a32454..04e92181b 100644 --- a/internal/api/handler_purchases_submit_dedupe_integration_test.go +++ b/internal/api/handler_purchases_submit_dedupe_integration_test.go @@ -19,27 +19,28 @@ import ( ) // insertBarrierStore holds each submit transaction at its INSERT until every -// participant has reached it (or the wait times out). That forces the -// interleaving MON-01 describes: both transactions have already run their -// duplicate check before either inserts. A guard that serializes the check -// (the per-creator lock) keeps the second transaction out of the barrier, so -// the first one times out and inserts alone. +// participant has either reached it or is parked on the per-creator submit +// lock. That forces the interleaving MON-01 describes: both transactions have +// already run their duplicate check before either inserts. A guard that +// serializes the check (the per-creator lock) keeps the second transaction +// waiting on that lock, which the barrier observes, so the first one inserts +// alone. The barrier never opens on a timeout: on a slow runner a fixed wait +// could let the first transaction insert and commit before an unguarded second +// one ran its check, hiding the missing guard. type insertBarrierStore struct { *config.PostgresStore + db testhelpers.LockDB // a pool connection other than the transactions' arrived chan struct{} parties int - wait time.Duration } +// submitLockWait matches ListRecentSubmitsTx's per-creator advisory lock. +const submitLockWait = "%pg_advisory_xact_lock%" + func (s *insertBarrierStore) SavePurchaseExecutionTx(ctx context.Context, tx pgx.Tx, e *config.PurchaseExecution) error { s.arrived <- struct{}{} - deadline := time.After(s.wait) - for len(s.arrived) < s.parties { - select { - case <-deadline: - return s.PostgresStore.SavePurchaseExecutionTx(ctx, tx, e) - case <-time.After(10 * time.Millisecond): - } + if err := testhelpers.AwaitArrivalsOrLockWaits(ctx, s.db, s.parties, func() int { return len(s.arrived) }, submitLockWait); err != nil { + return err } return s.PostgresStore.SavePurchaseExecutionTx(ctx, tx, e) } @@ -79,7 +80,7 @@ func setupSubmitDB(t *testing.T) (*testhelpers.PostgresContainer, *config.Postgr func TestPersistExecution_ConcurrentIdenticalSubmitsCreateOneRow(t *testing.T) { pg, store := setupSubmitDB(t) recs, key := submitFixture(t) - barrier := &insertBarrierStore{PostgresStore: store, arrived: make(chan struct{}, 2), parties: 2, wait: time.Second} + barrier := &insertBarrierStore{PostgresStore: store, db: pg.DB.Pool(), arrived: make(chan struct{}, 2), parties: 2} h := &Handler{config: barrier} var wg sync.WaitGroup From 52c7cbc105bc425a6b3adaba5303cf9d876545b2 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 29 Oct 2026 03:36:02 +0000 Subject: [PATCH 4/5] test(purchase): open the provider-lock race barriers on observed lock waits TestPlanProviderLock_AttachedNeverExceedsShown and TestAttachIntegration_PlansOnOneAccountNeverOverAttach held each tick at a barrier that gave up after 500ms. On a slow runner an unlocked tick could miss the barrier and see the first tick's committed suppression, hiding the over-commit the tests (and the lock-free control) depend on. The barrier now opens once every tick has resolved or is observed waiting on the per-provider advisory lock. A no-op LockPlanProvidersTx fails the attach test (12 attached against 10 shown). The locked runs no longer spend 500ms per waiting tick. Refs #36 Co-Authored-By: Claude Opus 5.5 --- .../plan_provider_lock_integration_test.go | 35 +++++++++++-------- .../plan_step_attach_integration_test.go | 18 ++++++---- 2 files changed, 32 insertions(+), 21 deletions(-) diff --git a/internal/purchase/plan_provider_lock_integration_test.go b/internal/purchase/plan_provider_lock_integration_test.go index 425311677..effcd134a 100644 --- a/internal/purchase/plan_provider_lock_integration_test.go +++ b/internal/purchase/plan_provider_lock_integration_test.go @@ -9,8 +9,10 @@ import ( "time" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" "github.com/google/uuid" "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" ) @@ -32,28 +34,33 @@ const ( planProviderWanted = 3 ) -func runProviderTicks(ctx context.Context, t *testing.T, store *config.PostgresStore, lock bool) int { +// planProviderLockWait matches LockPlanProvidersTx's advisory lock statement. +const planProviderLockWait = "%pg_advisory_xact_lock%" + +func runProviderTicks(ctx context.Context, t *testing.T, store *config.PostgresStore, pool *pgxpool.Pool, lock bool) int { t.Helper() account := uuid.NewString() var ( wg sync.WaitGroup mu sync.Mutex resolved int - allIn = make(chan struct{}) ) - // The barrier waits for every tick to have resolved, but gives up after a - // moment: with the lock the other ticks cannot resolve until this one - // commits, so they never arrive. + arrived := func() int { + mu.Lock() + defer mu.Unlock() + return resolved + } + // The barrier waits until every tick has resolved or is observed parked + // on the provider lock. With the lock the other ticks cannot resolve until + // this one commits; without it they all arrive. A fixed give-up timeout + // here let a slow unlocked tick miss the barrier, see the first tick's + // committed suppression, and hide the over-commit the control relies on. arrive := func() { mu.Lock() resolved++ - if resolved == planProviderTicks { - close(allIn) - } mu.Unlock() - select { - case <-allIn: - case <-time.After(500 * time.Millisecond): + if err := testhelpers.AwaitArrivalsOrLockWaits(ctx, pool, planProviderTicks, arrived, planProviderLockWait); err != nil { + t.Error(err) } } for i := 0; i < planProviderTicks; i++ { @@ -94,15 +101,15 @@ func runProviderTicks(ctx context.Context, t *testing.T, store *config.PostgresS func TestPlanProviderLock_AttachedNeverExceedsShown(t *testing.T) { ctx := context.Background() - store, _ := rampStepStore(ctx, t) + store, pool := rampStepStore(ctx, t) // Control: the harness really races. If this stops over-committing, the // assertion below proves nothing. - assert.Greater(t, runProviderTicks(ctx, t, store, false), planProviderCapacity, + assert.Greater(t, runProviderTicks(ctx, t, store, pool, false), planProviderCapacity, "without the lock the ticks must over-commit, otherwise the test cannot detect a missing lock") // Treatment: with the lock the ticks serialize and exactly fill the capacity. - assert.Equal(t, planProviderCapacity, runProviderTicks(ctx, t, store, true)) + assert.Equal(t, planProviderCapacity, runProviderTicks(ctx, t, store, pool, true)) } func TestLockPlanProvidersTx_Validation(t *testing.T) { diff --git a/internal/purchase/plan_step_attach_integration_test.go b/internal/purchase/plan_step_attach_integration_test.go index 1caf111ef..8909241f5 100644 --- a/internal/purchase/plan_step_attach_integration_test.go +++ b/internal/purchase/plan_step_attach_integration_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/LeanerCloud/cloud-commitments-platform/internal/config" + "github.com/LeanerCloud/cloud-commitments-platform/internal/database/postgres/testhelpers" "github.com/LeanerCloud/cloud-commitments-platform/internal/planstep" "github.com/jackc/pgx/v5/pgxpool" "github.com/stretchr/testify/assert" @@ -180,17 +181,20 @@ func TestAttachIntegration_PlansOnOneAccountNeverOverAttach(t *testing.T) { var mu sync.Mutex resolved := 0 - allIn := make(chan struct{}) + arrived := func() int { + mu.Lock() + defer mu.Unlock() + return resolved + } + // Opens once every tick has resolved or is observed parked on the + // per-provider lock (with the lock the others cannot resolve until this + // tick commits), never on a guessed timeout. barrier := func() { mu.Lock() resolved++ - if resolved == ticks { - close(allIn) - } mu.Unlock() - select { - case <-allIn: - case <-time.After(500 * time.Millisecond): // with the lock the others never arrive + if err := testhelpers.AwaitArrivalsOrLockWaits(ctx, pool, ticks, arrived, planProviderLockWait); err != nil { + t.Error(err) } } sender := newCountingSender(ctx) From b655a563056b1e8674502e70e088fad3bc878cb1 Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Thu, 29 Oct 2026 03:43:58 +0000 Subject: [PATCH 5/5] test: prove absences with synctest and Close ordering instead of quiet windows Four unit tests asserted that something did NOT happen by waiting a fixed time (200ms to 2s) and seeing nothing. A goroutine a loaded runner had not yet scheduled passed that window. - scheduler: the stale-hours-0 refresh check, the auth no-booking check on ValidateUserAPIKey and the server exactly-once usage booking checks now run in a synctest bubble. synctest.Wait returns only once every goroutine the call started has finished or is durably blocked, so a non-blocking receive afterwards is exact. Ignoring the disable sentinel, or booking usage on the validation path, fails every run (50/50). - server: the reinit cleanup-worker test replaced assert.Never(300ms) after canceling the request ctx with the positive half: Close must stop the worker while the request ctx is still live. Starting the worker with the request ctx (the #1137 call-site bug) fails it 10/10. The three usage-booking tests and the no-booking check drop from about 2s each to near zero. Closes #36 Co-Authored-By: Claude Opus 5.5 --- internal/auth/service_apikeys_test.go | 115 ++++++++++--------- internal/scheduler/scheduler_test.go | 24 ++-- internal/server/apikey_usage_booking_test.go | 115 +++++++++---------- internal/server/app_rl_cleanup_test.go | 25 ++-- 4 files changed, 152 insertions(+), 127 deletions(-) diff --git a/internal/auth/service_apikeys_test.go b/internal/auth/service_apikeys_test.go index d6044240e..2fbc3ac4a 100644 --- a/internal/auth/service_apikeys_test.go +++ b/internal/auth/service_apikeys_test.go @@ -8,6 +8,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/stretchr/testify/assert" @@ -659,66 +660,74 @@ func TestService_DeleteAPIKey(t *testing.T) { }) } -func TestService_ValidateUserAPIKey(t *testing.T) { - ctx := context.Background() +// validateBooksNoUsage is the "successfully validate API key" case. It must +// run inside a synctest bubble. +func validateBooksNoUsage(ctx context.Context, t *testing.T) { + mockStore := new(MockStore) + service := &Service{store: mockStore} - t.Run("successfully validate API key", func(t *testing.T) { - mockStore := new(MockStore) - service := &Service{store: mockStore} + apiKey := "test-api-key-123456" + hash := sha256.Sum256([]byte(apiKey)) + keyHash := base64.RawURLEncoding.EncodeToString(hash[:]) - apiKey := "test-api-key-123456" - hash := sha256.Sum256([]byte(apiKey)) - keyHash := base64.RawURLEncoding.EncodeToString(hash[:]) + user := &User{ + ID: "user-123", + Email: "test@example.com", + Active: true, + } - user := &User{ - ID: "user-123", - Email: "test@example.com", - Active: true, - } + apiKeyRecord := &UserAPIKey{ + ID: "key-1", + UserID: "user-123", + Name: "Test Key", + KeyHash: keyHash, + IsActive: true, + ExpiresAt: nil, + Permissions: []Permission{{Action: ActionView, Resource: ResourceRecommendations}}, + } - apiKeyRecord := &UserAPIKey{ - ID: "key-1", - UserID: "user-123", - Name: "Test Key", - KeyHash: keyHash, - IsActive: true, - ExpiresAt: nil, - Permissions: []Permission{{Action: ActionView, Resource: ResourceRecommendations}}, - } + mockStore.On("GetAPIKeyByHash", ctx, keyHash).Return(apiKeyRecord, nil) + mockStore.On("GetUserByID", ctx, "user-123").Return(user, nil) + // Validation must book NO usage: one HTTP request validates the same + // credential several times, so counting here would multiply every + // reported figure. The API layer books once per request instead -- + // see Handler.validateSecurityContext and the exactly-once tests in + // internal/server/apikey_usage_booking_test.go. + // + // Registered with .Maybe() rather than left unregistered because an + // unexpected call would panic inside the flush goroutine, where + // RecordUsageAsync's recover() would swallow it and the test would + // pass regardless. + usageRecorded := make(chan struct{}, 1) + mockStore.On("RecordAPIKeyUsage", mock.Anything, "key-1", mock.Anything). + Run(func(mock.Arguments) { usageRecorded <- struct{}{} }). + Return(nil).Maybe() - mockStore.On("GetAPIKeyByHash", ctx, keyHash).Return(apiKeyRecord, nil) - mockStore.On("GetUserByID", ctx, "user-123").Return(user, nil) - // Validation must book NO usage: one HTTP request validates the same - // credential several times, so counting here would multiply every - // reported figure. The API layer books once per request instead -- - // see Handler.validateSecurityContext and the exactly-once tests in - // internal/server/apikey_usage_booking_test.go. - // - // Registered with .Maybe() rather than left unregistered because an - // unexpected call would panic inside the flush goroutine, where - // RecordUsageAsync's recover() would swallow it and the test would - // pass regardless. - usageRecorded := make(chan struct{}, 1) - mockStore.On("RecordAPIKeyUsage", mock.Anything, "key-1", mock.Anything). - Run(func(mock.Arguments) { usageRecorded <- struct{}{} }). - Return(nil).Maybe() + resultKey, resultUser, err := service.ValidateUserAPIKey(ctx, apiKey) + + require.NoError(t, err) + assert.Equal(t, user, resultUser) + assert.Equal(t, apiKeyRecord, resultKey) + // A booking would be flushed by a goroutine started before + // ValidateUserAPIKey returned. synctest.Wait returns only once that + // goroutine has finished or is durably blocked, so by then any flush + // has already reached the store and the non-blocking receive is exact. + // A fixed quiet window could pass a flusher a loaded runner had not + // scheduled yet. + synctest.Wait() + select { + case <-usageRecorded: + t.Fatal("ValidateUserAPIKey must not record usage: booking on the validation path counts validations, not requests") + default: + } + mockStore.AssertExpectations(t) +} - resultKey, resultUser, err := service.ValidateUserAPIKey(ctx, apiKey) +func TestService_ValidateUserAPIKey(t *testing.T) { + ctx := context.Background() - require.NoError(t, err) - assert.Equal(t, user, resultUser) - assert.Equal(t, apiKeyRecord, resultKey) - // A booking would be flushed by a goroutine started before - // ValidateUserAPIKey returned, so a bounded quiet window is the only - // way to establish its absence. Unlike the positive waits elsewhere - // in this suite (where too short a deadline flakes), too short a - // window here can only miss a regression, never fail spuriously. - select { - case <-usageRecorded: - t.Fatal("ValidateUserAPIKey must not record usage: booking on the validation path counts validations, not requests") - case <-time.After(2 * time.Second): - } - mockStore.AssertExpectations(t) + t.Run("successfully validate API key", func(t *testing.T) { + synctest.Test(t, func(t *testing.T) { validateBooksNoUsage(ctx, t) }) }) t.Run("fail when owner's password changed after minting", func(t *testing.T) { diff --git a/internal/scheduler/scheduler_test.go b/internal/scheduler/scheduler_test.go index 2f911c52f..4a6f62ee1 100644 --- a/internal/scheduler/scheduler_test.go +++ b/internal/scheduler/scheduler_test.go @@ -10,6 +10,7 @@ import ( "sync" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/LeanerCloud/cloud-commitments-go/pkg/common" @@ -997,14 +998,23 @@ func staleListRecommendations(t *testing.T, staleHours int) <-chan struct{} { // store call is the second GetGlobalConfig in CollectRecommendations, so that // is what must never happen (MarkCollectionStarted belongs to the cold-start // path and says nothing about stale refreshes). +// +// The check runs in a synctest bubble: synctest.Wait returns only once every +// goroutine the call started has finished or is durably blocked, so a refresh +// it kicked off has already made that store call. The non-blocking receive is +// then a complete answer; a quiet window could pass a refresh goroutine that a +// loaded runner had not scheduled yet. func TestScheduler_ListRecommendations_StaleHoursZeroDisablesBackgroundRefresh(t *testing.T) { - refreshed := staleListRecommendations(t, 0) - - select { - case <-refreshed: - t.Fatal("background refresh started although RecommendationsCacheStaleHours == 0 disables it") - case <-time.After(200 * time.Millisecond): - } + synctest.Test(t, func(t *testing.T) { + refreshed := staleListRecommendations(t, 0) + synctest.Wait() + + select { + case <-refreshed: + t.Fatal("background refresh started although RecommendationsCacheStaleHours == 0 disables it") + default: + } + }) } // Positive control for the test above: with a finite TTL the same stale cache diff --git a/internal/server/apikey_usage_booking_test.go b/internal/server/apikey_usage_booking_test.go index c2ac1f673..8fd6e2c3f 100644 --- a/internal/server/apikey_usage_booking_test.go +++ b/internal/server/apikey_usage_booking_test.go @@ -6,6 +6,7 @@ import ( "encoding/base64" "sync/atomic" "testing" + "testing/synctest" "time" "github.com/LeanerCloud/cloud-commitments-platform/internal/api" @@ -142,42 +143,34 @@ func (h *usageBookingHarness) request(method, path string) *events.LambdaFunctio } // requireExactlyOneBooking asserts the request booked exactly one request -// against the key. +// against the key. It must run inside the synctest bubble that ran the +// request. // // The bookings themselves are synchronous -- RecordUsageAsync increments an // in-memory counter before returning -- so by the time HandleRequest returns, // every booking the request will ever make has already been made; only the -// flush to the store runs on a goroutine. With the bug present the surplus -// bookings therefore surface either as a first flush carrying a delta above 1 -// or as a second flush arriving microseconds later, and both are caught here. -// The second half needs a bounded quiet window rather than a channel receive -// because it is proving the ABSENCE of a further write. Note the two waits are -// not symmetric: the first is a positive wait, where too short a deadline -// causes a flaky FAILURE, so it is generous. The second is a negative wait, -// where too short a window can only cause a missed detection (a false PASS), -// never a flake -- and the surplus flush it is looking for comes from a -// goroutine already running before HandleRequest returned, so it lands within -// microseconds and the window is orders of magnitude wider than needed. +// flush to the store runs on a goroutine. synctest.Wait returns once that +// goroutine has finished (or is durably blocked), so every flush has reached +// the store and the drain below sees all of them. With the bug present the +// surplus bookings surface as a flush carrying a delta above 1 or as a second +// flush, and both are caught. A fixed quiet window could instead pass a +// second flush that a loaded runner had not scheduled yet. func (h *usageBookingHarness) requireExactlyOneBooking(t *testing.T) { t.Helper() - select { - case delta := <-h.flushes: - require.EqualValues(t, 1, delta, - "one HTTP request must book exactly 1 usage; the first flush wrote %d", delta) - case <-time.After(10 * time.Second): - t.Fatal("no API key usage reached the store") + synctest.Wait() + var deltas []int64 + for drained := false; !drained; { + select { + case delta := <-h.flushes: + deltas = append(deltas, delta) + default: + drained = true + } } - - select { - case delta := <-h.flushes: - t.Fatalf("a second flush booked %d more usage(s) (total %d): usage is being counted per credential validation, not per request", - delta, h.totalDelta.Load()) - case <-time.After(2 * time.Second): - } - - require.EqualValues(t, 1, h.totalDelta.Load(), - "total usage booked for one HTTP request") + require.Equal(t, []int64{1}, deltas, + "one HTTP request must book exactly 1 usage in one flush; flushes wrote %v (total %d): "+ + "usage is being counted per credential validation, not per request", deltas, h.totalDelta.Load()) } // TestHandleRequest_UserAPIKey_BooksOneUsagePerRequest pins the single-verb @@ -185,18 +178,20 @@ func (h *usageBookingHarness) requireExactlyOneBooking(t *testing.T) { // re-validates it inside the view:api-keys permission check, so booking on // the validation path reported one request as two. func TestHandleRequest_UserAPIKey_BooksOneUsagePerRequest(t *testing.T) { - h := newUsageBookingHarness(t, []auth.Permission{ - {Action: auth.ActionView, Resource: auth.ResourceAPIKeys}, + synctest.Test(t, func(t *testing.T) { + h := newUsageBookingHarness(t, []auth.Permission{ + {Action: auth.ActionView, Resource: auth.ResourceAPIKeys}, + }) + + resp, err := h.handler.HandleRequest(context.Background(), + h.request("GET", "/api/api-keys/usage-stats")) + + require.NoError(t, err) + require.Equal(t, 200, resp.StatusCode, "body: %s", resp.Body) + require.Greater(t, h.validations.Load(), int64(1), + "harness precondition: this endpoint must re-validate the credential, otherwise it cannot detect per-validation booking") + h.requireExactlyOneBooking(t) }) - - resp, err := h.handler.HandleRequest(context.Background(), - h.request("GET", "/api/api-keys/usage-stats")) - - require.NoError(t, err) - require.Equal(t, 200, resp.StatusCode, "body: %s", resp.Body) - require.Greater(t, h.validations.Load(), int64(1), - "harness precondition: this endpoint must re-validate the credential, otherwise it cannot detect per-validation booking") - h.requireExactlyOneBooking(t) } // TestHandleRequest_UserAPIKeyMultiVerbGate_BooksOneUsagePerRequest pins the @@ -208,18 +203,20 @@ func TestHandleRequest_UserAPIKey_BooksOneUsagePerRequest(t *testing.T) { // The resulting 403 is incidental -- what is pinned is that an authenticated // request books one usage no matter how many permission checks it triggers. func TestHandleRequest_UserAPIKeyMultiVerbGate_BooksOneUsagePerRequest(t *testing.T) { - h := newUsageBookingHarness(t, []auth.Permission{ - {Action: auth.ActionView, Resource: auth.ResourceRecommendations}, + synctest.Test(t, func(t *testing.T) { + h := newUsageBookingHarness(t, []auth.Permission{ + {Action: auth.ActionView, Resource: auth.ResourceRecommendations}, + }) + + resp, err := h.handler.HandleRequest(context.Background(), + h.request("DELETE", "/api/purchases/planned/11111111-1111-1111-1111-111111111111")) + + require.NoError(t, err) + require.Equal(t, 403, resp.StatusCode, "body: %s", resp.Body) + require.GreaterOrEqual(t, h.validations.Load(), int64(5), + "harness precondition: the four-verb gate must re-validate the credential once per verb") + h.requireExactlyOneBooking(t) }) - - resp, err := h.handler.HandleRequest(context.Background(), - h.request("DELETE", "/api/purchases/planned/11111111-1111-1111-1111-111111111111")) - - require.NoError(t, err) - require.Equal(t, 403, resp.StatusCode, "body: %s", resp.Body) - require.GreaterOrEqual(t, h.validations.Load(), int64(5), - "harness precondition: the four-verb gate must re-validate the credential once per verb") - h.requireExactlyOneBooking(t) } // TestHandleRequest_UserAPIKeyDeploymentInfo_BooksOneUsage pins issue #156: @@ -227,14 +224,16 @@ func TestHandleRequest_UserAPIKeyMultiVerbGate_BooksOneUsagePerRequest(t *testin // "/api/info" skipped the middleware's booking block for it. func TestHandleRequest_UserAPIKeyDeploymentInfo_BooksOneUsage(t *testing.T) { t.Setenv("CUDLY_SOURCE_CLOUD", "gcp") // skip the best-effort STS lookup - h := newUsageBookingHarness(t, []auth.Permission{ - {Action: auth.ActionView, Resource: auth.ResourceAPIKeys}, - }) + synctest.Test(t, func(t *testing.T) { + h := newUsageBookingHarness(t, []auth.Permission{ + {Action: auth.ActionView, Resource: auth.ResourceAPIKeys}, + }) - resp, err := h.handler.HandleRequest(context.Background(), - h.request("GET", "/api/info/deployment")) + resp, err := h.handler.HandleRequest(context.Background(), + h.request("GET", "/api/info/deployment")) - require.NoError(t, err) - require.Equal(t, 200, resp.StatusCode, "body: %s", resp.Body) - h.requireExactlyOneBooking(t) + require.NoError(t, err) + require.Equal(t, 200, resp.StatusCode, "body: %s", resp.Body) + h.requireExactlyOneBooking(t) + }) } diff --git a/internal/server/app_rl_cleanup_test.go b/internal/server/app_rl_cleanup_test.go index ac398dc06..6b9ca4bb6 100644 --- a/internal/server/app_rl_cleanup_test.go +++ b/internal/server/app_rl_cleanup_test.go @@ -89,25 +89,32 @@ func newReinitTestApp(t *testing.T) *Application { } // TestReinitializeAfterConnect_CleanupWorkerOutlivesRequestCtx pins the call -// site (#1137): the worker reinitializeAfterConnect starts must not stop when -// the request context that triggered the lazy DB connect ends (#269), and must -// stop on Close. Starting it with dbRL.StartCleanupWorker(ctx) fails this. +// site (#1137): the worker reinitializeAfterConnect starts must run on the app +// lifetime ctx, not on the request context that triggered the lazy DB connect +// (#269), so Close must stop it while that request ctx is still live. Starting +// it with dbRL.StartCleanupWorker(ctx) leaves it running after Close and fails +// this. +// +// This checks the positive half (Close stops it) instead of a quiet window +// after canceling the request ctx: such a window passes a request-bound worker +// that a loaded runner had not yet scheduled to exit. The other half is +// structural: startRateLimiterCleanup takes no ctx, and +// TestStartRateLimiterCleanup_SurvivesRequestCtxAndStopsOnClose pins the ctx +// it hands the worker. func TestReinitializeAfterConnect_CleanupWorkerOutlivesRequestCtx(t *testing.T) { app := newReinitTestApp(t) before := liveDBRLWorkers() reqCtx, cancelReq := context.WithCancel(testutil.TestContext(t)) + defer cancelReq() // only after the assertions: the ctx stays live throughout require.NoError(t, app.reinitializeAfterConnect(reqCtx, &database.Connection{})) require.Equal(t, before+1, liveDBRLWorkers(), "reinitializeAfterConnect must start one cleanup worker") - cancelReq() - assert.Never(t, func() bool { return liveDBRLWorkers() <= before }, - 300*time.Millisecond, 10*time.Millisecond, - "cleanup worker stopped with the request ctx; it must run on the app lifetime ctx") - require.NoError(t, app.Close()) assert.Eventually(t, func() bool { return liveDBRLWorkers() == before }, - 2*time.Second, 10*time.Millisecond, "cleanup worker must stop on Close") + 2*time.Second, 10*time.Millisecond, + "cleanup worker must stop on Close while the request ctx is still live; it must run on the app lifetime ctx") + require.NoError(t, reqCtx.Err(), "the request ctx must still be live when Close stops the worker") } // TestReinitializeAfterConnect_NoCleanupWorkerWhenLaterStepFails: ensureDB