Transactions & Pipelines

Two distinct tools that are often confused:

• $transaction, ACID atomicity across multiple statements. One BEGIN / COMMIT pair, rollback on throw.
• pipeline, throughput. N independent queries sent in one TCP round-trip via the Postgres extended-query protocol. Not atomic.

Use transactions when you need all-or-nothing semantics. Use pipelines when you need latency.

$transaction, callback form#

Pass a callback that receives a transactional client. Throw to roll back, return to commit.

const result = await db.$transaction(async (tx) => {
  const user = await tx.users.create({
    data: { email: 'alice@example.com', name: 'Alice' },
  });
 
  await tx.posts.create({
    data: { userId: user.id, title: 'Hello' },
  });
 
  return user;
});
// result is the User returned by the callback

The tx client has the same surface as db, every table accessor, every method. Queries must use tx, not db, or they'll run on a different connection outside the transaction.

Note: If the opening BEGIN itself fails, Turbine no longer fires a best-effort ROLLBACK afterward (fixed in v0.30). On single-handle engines like embedded PowDB, that stray rollback could tear down an unrelated open transaction; skipping it keeps a failed BEGIN from corrupting other work.

$transaction, batch form#

Pass an array of DeferredQuery objects (the build* methods, same as pipeline) and $transaction runs them atomically on one connection, BEGIN, each query in order, COMMIT, returning a positionally-typed tuple of each query's result. Any error rolls the whole batch back and rethrows. Unlike a pipeline, this is atomic.

const [user, postCount] = await db.$transaction([
  db.users.buildFindUnique({ where: { id: 1 } }),
  db.posts.buildCount({ where: { userId: 1 } }),
]);
// user: User | null, postCount: number, all-or-nothing

Use the batch form when you have a fixed set of independent statements that must succeed or fail together and you don't need a callback's control flow. An empty array resolves to [] without opening a transaction. The callback form above is unchanged, $transaction accepts either.

Batch form vs pipeline: both take DeferredQuery[]. The pipeline optimizes for latency (one round-trip, not atomic); the batch $transaction optimizes for atomicity (BEGIN…COMMIT, rollback on any failure).

Isolation levels#

Pass isolationLevel to set the transaction's isolation on BEGIN:

await db.$transaction(
  async (tx) => {
    const row = await tx.inventory.findUnique({ where: { sku: 'ABC' } });
    await tx.inventory.update({
      where: { sku: 'ABC' },
      data: { stock: { decrement: 1 } },
    });
  },
  { isolationLevel: 'Serializable' },
);

Supported: 'ReadUncommitted', 'ReadCommitted' (Postgres default), 'RepeatableRead', 'Serializable'.

When Serializable detects a conflict, it throws SerializationFailureError (TURBINE_E013) with isRetryable: true as const. See the retry section below.

Timeouts#

Per-transaction timeout in milliseconds, enforced client-side: if the callback exceeds timeout, Turbine destroys the connection (aborting any in-flight statement server-side) and rolls back:

await db.$transaction(
  async (tx) => {
    await tx.reports.create({ data: { /* ... */ } });
  },
  { timeout: 5_000 },
);

A query that exceeds the timeout throws TimeoutError (TURBINE_E002) and the transaction rolls back.

Nested transactions, real SAVEPOINTs#

Calling $transaction inside another $transaction opens a Postgres SAVEPOINT, not a no-op and not a new connection. The inner block can fail and recover independently:

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 {
    // Inner block rolled back to SAVEPOINT; outer is still healthy
  }
 
  await tx.posts.create({ data: { userId: user.id, title: 'Hello' } });
  // user + post still commit
});

SAVEPOINT names are auto-generated with a counter so nesting is safe.

Multi-tenant queries with RLS session context#

Set transaction-local Postgres settings (GUCs) so PostgreSQL Row-Level Security policies that call current_setting() filter rows for you. This is the clean way to do multi-tenant isolation: the database enforces the tenant boundary, not your application code.

Pass sessionContext to $transaction:

// Postgres policy: USING (tenant_id = current_setting('app.current_tenant')::int)
const rows = await db.$transaction(
  async (tx) => tx.documents.findMany(),
  { sessionContext: { 'app.current_tenant': tenantId } },
);
// Only this tenant's documents come back, the policy did the filtering

Each entry is applied as SELECT set_config(name, value, true) right after BEGIN, so the setting is scoped to the transaction and resets automatically on commit or rollback. Values may be strings, numbers, or booleans (numbers and booleans are coerced to strings, since GUCs are text). An invalid setting name throws ValidationError and rolls the transaction back before any query runs.

$withSession shorthand#

For a single-purpose session, $withSession skips the options object, pass the context first, then the callback:

const rows = await db.$withSession(
  { 'app.current_tenant': tenantId },
  async (tx) => tx.documents.findMany(),
);

$withSession(ctx, fn) is exactly $transaction(fn, { sessionContext: ctx }). Use it when the only reason you're opening a transaction is to scope the RLS context.

Retryable errors#

Two error classes carry readonly isRetryable = true as const, so TypeScript narrows them and you don't need a runtime field check.

Canonical retry loop, using the withRetry helper the package exports (or the identical db.$retry):

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

withRetry retries only errors carrying isRetryable === true and rethrows everything else at once. Tune it with maxAttempts (default 3), baseDelay (default 50 ms), maxDelay (default 5000 ms), and onRetry(error, attempt); backoff is exponential with jitter.

The as const on isRetryable means err.isRetryable is the literal type true on those classes and never narrows to false, so your control flow type-checks without casts, whether you use the helper or write your own loop.

pipeline, N queries, 1 round-trip#

A pipeline bundles independent queries and sends them together using the Postgres extended-query pipeline protocol. Each query gets its own parsed SQL, own params, own typed result. No transactional semantics, if query 3 fails, queries 1, 2, 4, 5 still ran.

const [user, posts, commentCount, orgs, latestLogin] = await db.pipeline([
  db.users.buildFindUnique({ where: { id: 1 } }),
  db.posts.buildFindMany({ where: { userId: 1 }, limit: 10 }),
  db.comments.buildCount({ where: { userId: 1 } }),
  db.organizations.buildFindMany({ where: { ownerId: 1 } }),
  db.sessions.buildFindFirst({ where: { userId: 1 }, orderBy: { createdAt: 'desc' } }),
]);

Each build* method returns a DeferredQuery<T>, { sql, params, transform, tag }. The pipeline driver writes them all to the wire, reads all responses, applies each transform.

Result type is a tuple that matches the input order and carries the per-query return type.

Driving a pipeline yourself#

db.pipeline() is the ergonomic wrapper. The underlying function is exported too, for cases where you hold a pg.Pool directly rather than a TurbineClient:

import { executePipeline, pipelineSupported } from 'turbine-orm';
 
const results = await executePipeline(pool, queries, options);

executePipeline(pool, queries, options?) takes a pg.Pool, a readonly array of DeferredQuery objects, and the same PipelineOptions as db.pipeline(); it returns the same order-matched tuple. An empty array short-circuits to [] without touching the pool. await pipelineSupported(pool) resolves to a boolean telling you whether that pool's driver exposes the wire-protocol path (some pg-compatible serverless drivers do not), so you can fall back to sequential execution rather than fail.

When pipeline wins#

• Edge runtimes. One round-trip to Neon (~35 ms) instead of five (~175 ms).
• Dashboard loads. Five independent widget queries → one round-trip.
• Fan-out reads. Load the user + their N related collections for a profile page.

When pipeline loses#

• You need atomicity. Use $transaction.
• One query depends on another's result. Pipeline queries must be independent, they're all submitted before any response is read.
• A query might fail. PipelineError (TURBINE_E014) wraps partial-success cases; decide how you want to handle them.

Batching inside a transaction#

There is no tx.pipeline(). The extended-query pipeline protocol needs to own the connection, and inside a transaction the connection is already owned by the BEGIN block, so the two cannot be combined. If you want a set of statements to run atomically as one unit, use the array form of $transaction at the top level instead:

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

That runs every statement on one connection inside one BEGIN / COMMIT, and rolls the whole batch back on the first failure. It is sequential on the connection by default, which is safe on every driver including HTTP and serverless pools. The one-write-burst path needs both conditions to hold: the checked-out connection advertises supportsPipelining, and the active dialect's resultStrategy is not 'reselect'. When both are true the statements are dispatched in one write burst with replies collected in order, saving a round trip each. MySQL is 'reselect', so the burst never happens there and the batch always runs sequentially. Either way this is a transaction, not the pipeline protocol: for that, use db.pipeline(...) outside a transaction.

TransactionClient reference#

The tx handed to your $transaction callback is a TransactionClient, and it is deliberately a smaller object than db. Everything it exposes runs on the single dedicated connection the transaction holds.

What tx has#

The one internal member#

tx.rawQuery(text, params) is marked @internal and is not application API. It is the seam turbine-orm/prisma-compat detects by shape (typeof tx.rawQuery === 'function') so it can run compat raw SQL on the transaction's own connection instead of quietly falling back to a pool connection outside the transaction. It takes the SQL as a plain string rather than a tagged template, which moves the escaping discipline to the caller, and that is precisely the property tx.raw exists to remove. It still exists at runtime and still typechecks, so code already calling it keeps working; it is simply no longer advertised. Use tx.raw.

What tx does not have#

Reaching for any of these is a compile error (TS2339: Property '<x>' does not exist on type 'TransactionClient'). TransactionClient is a plain class whose declared members are table(), $transaction(), raw, schema and the internal rawQuery() above, with no index signature, so TypeScript rejects every name below before the code ever runs. The table accessors are installed at construction with Object.defineProperty and typed by the generated TypedTransactionClient subclass, which declares one property per table. Neither the base class nor the generated subclass has a catch-all index signature, so nothing below type-checks.

Because every one of these is caught at compile time, the practical failure mode is a red squiggle in your editor, not a production TypeError. If you are reaching for one of them, the "use instead" column is the actual answer.

Transaction options#

Options are passed as the second argument to db.$transaction(fn, options) and apply to the whole block.

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

• isolationLevel: one of 'ReadUncommitted', 'ReadCommitted', 'RepeatableRead', 'Serializable'. Composed into the BEGIN statement by the active dialect. Under Serializable or RepeatableRead, expect SerializationFailureError (TURBINE_E013) and retry the whole block.
• timeout: milliseconds. When it fires, the transaction is rolled back and the connection is destroyed rather than returned to the pool, so the in-flight backend query is actually aborted rather than left running. You get a TimeoutError (TURBINE_E002).
• sessionContext: a record of transaction-local Postgres GUCs, applied as SELECT set_config($1, $2, true) immediately after BEGIN and before your callback. Both the name and the value are bound parameters, and the name is additionally checked against a strict identifier pattern, so a malformed name throws ValidationError and rolls back before any query runs. is_local = true means the setting resets on COMMIT or ROLLBACK and never leaks onto the pooled connection. This is Postgres-only: on an engine whose dialect reports no RLS support it throws UnsupportedFeatureError (TURBINE_E017). db.$withSession(context, fn) is the shorthand for db.$transaction(fn, { sessionContext: context }).

The raw transaction API#

db.transaction(fn) (no $) is the older, lower-level form: your callback receives a raw pg.PoolClient rather than a TransactionClient, with the same BEGIN / COMMIT / rollback-on-throw handling and no typed table accessors. Prefer $transaction; reach for this only when you want to issue hand-written SQL on the transaction's own connection.

See also#

• Relations, reads that fan out via with vs pipeline.
• Typed Errors, DeadlockError, SerializationFailureError, TimeoutError, PipelineError.
• Serverless, why pipelines matter more on the edge.