diff --git a/packages/services/src/monitor/__tests__/stream-monitor-preview.test.ts b/packages/services/src/monitor/__tests__/stream-monitor-preview.test.ts index dcb211dca..f2e7c1924 100644 --- a/packages/services/src/monitor/__tests__/stream-monitor-preview.test.ts +++ b/packages/services/src/monitor/__tests__/stream-monitor-preview.test.ts @@ -55,8 +55,8 @@ describe("streamMonitorPreview", () => { }); test("throws NotFoundError when monitor belongs to a different workspace", async () => { + const ws = (await createWorkspaceFixture("team")).workspace; await withTestTransaction(async (tx) => { - const ws = (await createWorkspaceFixture("team")).workspace; const ctx = { ...makeUserCtx(ws), db: tx }; // Use a monitor id that does not exist for this workspace. @@ -76,8 +76,8 @@ describe("streamMonitorPreview", () => { }); test("yields one result per region for an owned monitor", async () => { + const ws = (await createWorkspaceFixture("team")).workspace; await withTestTransaction(async (tx) => { - const ws = (await createWorkspaceFixture("team")).workspace; const ctx = { ...makeUserCtx(ws), db: tx }; const created = await createMonitor({ ctx, diff --git a/packages/services/src/page-subscriber/__tests__/page-subscriber.test.ts b/packages/services/src/page-subscriber/__tests__/page-subscriber.test.ts index 563db004f..15c490d1d 100644 --- a/packages/services/src/page-subscriber/__tests__/page-subscriber.test.ts +++ b/packages/services/src/page-subscriber/__tests__/page-subscriber.test.ts @@ -1028,8 +1028,8 @@ describe("hasPendingSubscriber", () => { describe("createPageSubscriber", () => { test("rejects read-only actor", async () => { + const team = (await createWorkspaceFixture("team")).workspace; await withTestTransaction(async (tx) => { - const team = (await createWorkspaceFixture("team")).workspace; const readOnlyCtx = { ...makeApiKeyCtx(team, { keyId: "k-read", diff --git a/packages/services/src/sso/__tests__/sso.test.ts b/packages/services/src/sso/__tests__/sso.test.ts index 2ad919f8a..31aa1fa35 100644 --- a/packages/services/src/sso/__tests__/sso.test.ts +++ b/packages/services/src/sso/__tests__/sso.test.ts @@ -715,10 +715,10 @@ describe("joinSsoWorkspace", () => { } test("provisions the member and audits the join", async () => { + // `createUser` writes on the default db — do it before the test tx takes + // the write lock, or the two deadlock. + const newcomer = await createUser(); await withTestTransaction(async (tx) => { - // `createUser` writes on the default db — do it before the test tx takes - // a write lock, or the two deadlock. - const newcomer = await createUser(); await enableSsoOn(tx); const ctx = makeUserCtx(scaleCtx.workspace, { userId: newcomer.id }); @@ -774,8 +774,8 @@ describe("joinSsoWorkspace", () => { }); test("rejects a workspace with sso switched off", async () => { + const newcomer = await createUser(); await withTestTransaction(async (tx) => { - const newcomer = await createUser(); await expect( joinSsoWorkspace({ ctx: { ...scaleCtx, db: tx }, @@ -786,8 +786,8 @@ describe("joinSsoWorkspace", () => { }); test("a sign-in that overlaps disableSso does not provision", async () => { + const newcomer = await createUser(); await withTestTransaction(async (tx) => { - const newcomer = await createUser(); await enableSsoOn(tx); // The ctx snapshot was taken while SSO was on; the owner turns it off // before the callback lands. diff --git a/packages/services/test/helpers.ts b/packages/services/test/helpers.ts index 584359bd7..cdad20e86 100644 --- a/packages/services/test/helpers.ts +++ b/packages/services/test/helpers.ts @@ -1,4 +1,4 @@ -import { and, db, desc, eq, inArray } from "@openstatus/db"; +import { and, db, desc, eq, inArray, sql } from "@openstatus/db"; import { type Scope, auditLog, @@ -8,10 +8,41 @@ import { createTestWorkspace } from "@openstatus/db/src/test/factories"; import { expect } from "vitest"; import type { Actor, DB, DrizzleTx, ServiceContext } from "../src/context"; +import { isRetryableDbError } from "../src/retry"; import type { Workspace } from "../src/types"; class RollbackSignal extends Error {} +/** The transaction never started: `fn` did not run, so it's safe to retry. */ +class BeginFailed extends Error { + constructor(readonly original: unknown) { + super("test transaction failed to begin"); + } +} + +// `db.transaction()` opens with `BEGIN IMMEDIATE`, so every test transaction +// takes the single write lock up front. Parallel test files queue on it, and a +// waiter that loses gets SQLITE_BUSY ("database is locked") or sqld's +// TRANSACTION_TIMEOUT — surfaced on the test's first query, since libsql +// pipelines BEGIN with it. +const BEGIN_MAX_ATTEMPTS = 8; +const BEGIN_BASE_DELAY_MS = 50; +const BEGIN_MAX_DELAY_MS = 1_000; + +function isLockContention(err: unknown): boolean { + if (isRetryableDbError(err)) return true; + let current: unknown = err; + for (let depth = 0; depth < 10; depth++) { + if (typeof current !== "object" || current === null) return false; + if ("code" in current && current.code === "TRANSACTION_TIMEOUT") { + return true; + } + if (!("cause" in current) || current.cause === current) return false; + current = current.cause; + } + return false; +} + /** * Wrap a test body in a transaction that always rolls back, so per-test * writes (rows, audit log entries, etc.) never reach the committed db. @@ -30,13 +61,45 @@ class RollbackSignal extends Error {} * - Inside the callback, replace any `db.select(...)` / * `db.insert(...)` / `db.delete(...)` with the same call on `tx`. * Reads via committed `db` will not see the in-tx writes. + * + * Lock contention while BEGIN acquires the write lock is retried with + * backoff. Once the transaction has started, `fn` holds the lock, so its + * own queries are never retried (and `fn` never runs twice). */ export async function withTestTransaction( fn: (tx: DrizzleTx) => Promise, +): Promise { + for (let attempt = 1; ; attempt++) { + try { + return await runRolledBack(fn); + } catch (e) { + if (!(e instanceof BeginFailed)) throw e; + if (attempt >= BEGIN_MAX_ATTEMPTS || !isLockContention(e.original)) { + throw e.original; + } + const delay = Math.min( + BEGIN_BASE_DELAY_MS * 2 ** (attempt - 1), + BEGIN_MAX_DELAY_MS, + ); + await new Promise((resolve) => + setTimeout(resolve, delay / 2 + Math.random() * (delay / 2)), + ); + } + } +} + +async function runRolledBack( + fn: (tx: DrizzleTx) => Promise, ): Promise { let result!: T; try { await db.transaction(async (tx) => { + // Force the BEGIN now so lock contention is isolated from `fn`. + try { + await tx.run(sql`select 1`); + } catch (e) { + throw new BeginFailed(e); + } result = await fn(tx); throw new RollbackSignal(); });