Transactions & Pipelines

Turbine has two ways to run several statements together:

• $transaction makes them atomic: one BEGIN and COMMIT, rolled back if your callback throws.
• pipeline cuts latency: independent queries in one round trip, in one transaction by default.

$transaction, callback form#

Return to commit, throw to roll back:

const user = await db.$transaction(async (tx) => {
  const created = await tx.users.create({ data: { email: 'alice@example.com' } });
  await tx.posts.create({ data: { userId: created.id, title: 'Hello' } });
  return created;
});

Run every query through tx. A query through db runs on another connection, outside the transaction.

Every query finishes inside the callback#

Await everything before the callback returns. If a query on tx is still running when it returns, Turbine rolls back and rejects with ValidationError (E003). A query on tx after the transaction settles throws the same error.

await db.$transaction(async (tx) => {
  // ❌ forEach does not wait, so the transaction rolls back
  ids.forEach(async (id) => {
    await tx.posts.update({ where: { id }, data: { published: true } });
  });
 
  // ✅
  for (const id of ids) {
    await tx.posts.update({ where: { id }, data: { published: true } });
  }
});

Transaction options#

Pass options as the second argument:

await db.$transaction(async (tx) => tx.invoices.findMany(), {
  isolationLevel: 'Serializable',
  timeout: 5_000,
});

Pair readOnly with isolationLevel: 'RepeatableRead' when several reads must see one snapshot.

Row-Level Security with sessionContext#

// Policy: USING (tenant_id = current_setting('app.current_tenant')::int)
const docs = await db.$transaction(async (tx) => tx.documents.findMany(), {
  sessionContext: { 'app.current_tenant': tenantId },
});
 
// The same thing, shorter
const docs2 = await db.$withSession({ 'app.current_tenant': tenantId }, (tx) =>
  tx.documents.findMany(),
);

• Each entry runs as set_config(name, value, true), so it resets at commit or rollback.
• Values must be strings, numbers or booleans. undefined throws E003 before a connection is taken.
• Names need a dot. A built-in setting such as role throws E003 unless you list it:

await db.$transaction(async (tx) => tx.documents.findMany(), {
  sessionContext: { 'app.current_tenant': tenantId, statement_timeout: '5s' },
  allowBuiltinSettings: ['statement_timeout'],
});

List only names your code sets, never one from a request. sessionContext is PostgreSQL only; the RLS recipe covers the policy side.

Nested transactions (savepoints)#

Calling $transaction on tx opens a SAVEPOINT. If the inner block throws, only it rolls back:

await db.$transaction(async (tx) => {
  const user = await tx.users.create({ data: { email: 'a@b.c' } });
 
  try {
    await tx.$transaction(async (inner) => {
      await inner.auditLog.create({ data: { userId: user.id, action: 'signup' } });
      throw new Error('skip audit log');
    });
  } catch {
    // only the savepoint rolled back
  }
 
  await tx.posts.create({ data: { userId: user.id, title: 'Hello' } }); // still commits
});

• Use the inner handle (inner) for the block's work. A query through the outer tx while the block is open throws E003.
• tx.$transaction takes no options. Isolation, timeout and session context belong to the outer transaction.
• Blocks that share a parent run one at a time, even under Promise.all.
• A nested create, nested update or lookup-first upsert on tx gets its own savepoint, so a failure part way leaves the transaction as it was.

A caught error still rolls the transaction back#

PostgreSQL aborts a transaction at its first failed statement, and catching the error does not undo that. Nothing is saved. $transaction rejects with TransactionRolledBackError (E019), with the first failure as cause, and later statements are refused without being sent. A deadlock, serialization failure or unavailable lock keeps its own retryable class.

To survive an expected failure, run that statement in a nested block:

import { UniqueConstraintError } from 'turbine-orm';
 
await db.$transaction(async (tx) => {
  const user = await tx.users.create({ data: { email } });
  try {
    await tx.$transaction((inner) =>
      inner.subscriptions.create({ data: { userId: user.id, plan: 'trial' } }),
    );
  } catch (err) {
    if (!(err instanceof UniqueConstraintError)) throw err;
  }
  // the user row still commits
});

Timeouts#

When timeout fires, Turbine cancels the running statement and rolls back. Only then does the call reject, with TimeoutError (E002) and outcome: 'not-executed', so nothing was saved and the locks are free:

import { TimeoutError } from 'turbine-orm';
 
try {
  await db.$transaction(async (tx) => tx.reports.create({ data: report }), { timeout: 5_000 });
} catch (err) {
  if (!(err instanceof TimeoutError)) throw err;
  err.outcome; // 'not-executed': nothing was saved, so a retry is safe
}

timeout takes milliseconds below 2^31, or 0 for none. Anything else throws E003.

On PostgreSQL, a per-query timeout inside a transaction cancels that statement and marks the transaction rolled back, so returning afterwards rejects with E019. Run the statement in a nested block to keep going.

Retryable errors#

Each has isRetryable: true. Retry the whole transaction with withRetry (or db.$retry):

import { withRetry } from 'turbine-orm';
 
await withRetry(
  () => db.$transaction(async (tx) => { /* ... */ }, { isolationLevel: 'Serializable' }),
  { maxAttempts: 5 },
);

Any other error is rethrown at once.

Lost connections#

A restart, failover or suspended serverless compute closes connections from the server side. The pending call rejects with ConnectionError (E004), and the process keeps running. Turbine resends only what cannot double-apply:

A lost write may have committed before the reply was lost, so retry it yourself only if it is idempotent. Pools Turbine opens also evict connections that died while a serverless function was frozen, before lending them out.

$transaction, batch form#

Pass build* queries instead of a callback. They run in order in one transaction and return a typed tuple:

const [user, postCount] = await db.$transaction([
  db.users.buildFindUnique({ where: { id: 1 } }),
  db.posts.buildCount({ where: { userId: 1 } }),
]);

Items must be build* queries. A promise such as db.users.create(...) already started outside the transaction, so it throws E003 before anything runs. The batch form takes the same options as the callback form.

Pipeline, N queries, 1 round trip#

const [user, posts, commentCount] = await db.pipeline([
  db.users.buildFindUnique({ where: { id: 1 } }),
  db.posts.buildFindMany({ where: { userId: 1 }, limit: 10 }),
  db.comments.buildCount({ where: { userId: 1 } }),
]);

A pipeline sends every query before reading any reply, so use it for independent queries. By default the batch runs in one transaction, and any failure rolls it all back. With { transactional: false } each query stands alone, and a partial failure throws PipelineError (E014):

import { PipelineError } from 'turbine-orm';
 
try {
  await db.pipeline(queries, { transactional: false });
} catch (err) {
  if (err instanceof PipelineError) {
    err.results; // [{ status: 'ok', value }, { status: 'error', error }, ...]
  }
}

Without the pipeline protocol (some serverless drivers, and the non-PostgreSQL engines), the queries run one after another on one connection. await db.pipelineSupported() tells you which path you get. With a raw pg.Pool, executePipeline(pool, queries, options) and pipelineSupported(pool) do the same.

There is no tx.pipeline(). For an atomic batch, use the batch form of db.$transaction.

What tx has#

await db.$transaction(async (tx) => {
  const rows = await tx.raw<{ n: number }>`
    SELECT count(*)::int AS n FROM posts WHERE user_id = ${userId}`;
  await tx.users.update({ where: { id: userId }, data: { postCount: rows[0].n } });
});

Raw SQL inside a transaction runs no $use middleware. It also refuses BEGIN, COMMIT, ROLLBACK, SAVEPOINT and other transaction-control statements with E003: return to commit, throw to roll back.

tx has no pipeline, $use, $listen, $on, $retry, $withSession or pool methods. Use those on db, and wrap the whole db.$transaction(...) call to retry it.

The raw transaction API#

db.transaction(fn) (no $) hands your callback the raw driver client instead of tx, with the same commit and rollback handling. Use it only for hand-written SQL:

await db.transaction(async (client) => {
  await client.query('UPDATE accounts SET balance = balance - $1 WHERE id = $2', [100, from]);
  await client.query('UPDATE accounts SET balance = balance + $1 WHERE id = $2', [100, to]);
});

SAVEPOINT, RELEASE and ROLLBACK TO work on it. BEGIN, COMMIT, ROLLBACK and release() throw E003.

See also#

• Relations: reads that fan out with with.
• Recipes: Row-Level Security end to end.
• Typed errors: every error named here.
• Serverless: why pipelines matter more on the edge.