TransactionsCopy page

tx() runs a callback on one reserved connection of the primary. The callback's client is the one to use. tx() commits when the callback resolves and rolls back when it throws. tx() returns what the callback returns. batch() runs write handles as one atomic unit on the primary, and returns one result per handle, in order.

Call

tx(fn) and tx(options, fn). Options go first. tx(fn, options) is OKM1120: options go first: tx(options, fn). A missing callback is OKM1121: tx expects a callback: tx(fn) or tx(options, fn). Options that are not an object are the same code. An unknown key is OKM1120 and lists isolation, retry, signal, and timeout.

After the callback returns, the connection is released. A later call on that client is The transaction has finished. Use the client the callback received only while the callback runs.

Options

Another isolation is OKM1120: isolation must be one of: read committed, repeatable read, serializable. A retry that is not an integer from 0 up, and a timeout that is not a number above 0, are OKM1121.

Retry

retry runs the callback again only when the error kind is serialization or deadlock. Each run is a new transaction, after a short wait. The deadline is set once. When it has passed, the call fails as kind timeout: The transaction timed out. An aborted signal fails as cancelled and stops the loop.

OKM1401 is the unknown outcome. COMMIT was sent and the result never arrived, so the transaction may or may not have committed. On tx() the message is The commit was sent and its result never arrived. The transaction may or may not have committed. A lost COMMIT of a batch() is the same code and the same kind, outcome_unknown. That call is never retried. Check whether the write landed before you run it again.

A constraint failure, a timeout, a cancel, and every other kind are thrown as they are.

Nested

A tx() inside a transaction is a savepoint on the same connection. It accepts no options. Any of isolation, retry, timeout, or signal is OKM1121: isolation, retry, timeout, and signal belong to the outermost tx(). A nested tx() is a savepoint on the same transaction.

The savepoint is released when the callback resolves. When the callback throws, the savepoint rolls back. If the caller catches that error, the outer transaction continues. The outer deadline still bounds the savepoint. That ceiling is Pool.

afterCommit

afterCommit registers a function on the transaction client. The functions run in registration order after the outermost transaction commits, and before tx() resolves. A function registered on a savepoint joins that list only when the savepoint is released. A rollback of the outermost transaction runs none of them.

An error in a function is reported to hookm.onError and the next function still runs. The value tx() returns stays the callback's value. Observers are Observability. Passing a value that is not a function is OKM1121: afterCommit expects a function.

batch

batch() takes write handles and runs them as one unit on the primary. Pass insert, update, delete, and archive without awaiting them. A failure rolls the unit back. batchIndex is the operation that failed, or null when the failure is at commit.

tx(fn) gives you a callback between statements, plus isolation and retry. batch() takes signal and timeout. Inside tx(), the unit is a savepoint. The error reaches the callback. Handling it leaves the outer transaction open. Letting it throw rolls the outer transaction back.

.replica() is OKM1840. restore, a write with expect, refresh(), an update that sets a path column, and an update that sets a tree parent are OKM1121. A versioned update is OKM1121 on db.batch. The same update is accepted inside tx().

One connection

tx() holds one connection from BEGIN until commit, rollback, or cancellation, then returns it to the pool. With replicas, reserve takes the primary. The route reason is primary-required. A replica route inside the callback, and using("replica"), are OKM1840: This operation needs the primary. Routing is Replicas.

Drivers

tx() needs transactions: "interactive". Neon over HTTP is transactions: "batch", so tx() with no timeout is OKM1111: tx() needs a driver with transactions: "interactive". This driver has "batch". The fix names batch(). WebSocket mode is interactive, so tx() runs. The suite is Compatibility. Driver setup is Drivers.

A positive timeout is OKM1120, and the message is timeout, when the driver cannot cancel. That check runs before the capability check. It covers Bun.sql, PGlite, and Neon, on a call, on tx(), and on batch(). postgres.js and node-postgres accept a positive timeout. On PGlite, a call made on the outer client from inside a tx() callback queues behind the transaction; use the client the callback received.

Timeouts

timeout on tx() bounds the whole call, including retries. Omit it and the bound is timeouts.transaction. timeouts.idleInTransaction bounds the gap between statements. The keys and the messages are Pool.

A transfer

The callback moves 40 from ada to grace, then commits. afterCommit prints after the commit.

import { connect } from "okmodel/pg/postgresjs";
import { schema, table, t } from "okmodel/pg";

const accounts = table("accounts", {
  id: t.text().primaryKey(),
  balance: t.integer(),
});
export const app = schema({ tables: [accounts] });

export async function transfer(url: string): Promise<void> {
  await using db = connect(url, { schema: app });
  await db.accounts.insert([
    { id: "ada", balance: 100 },
    { id: "grace", balance: 50 },
  ]);
  await db.tx(async (tx) => {
    tx.afterCommit(() => {
      console.log("committed"); 
    });
    const [from] = await tx.accounts.find({ where: { id: "ada" }, limit: 1 });
    const [to] = await tx.accounts.find({ where: { id: "grace" }, limit: 1 });
    if (from === undefined || to === undefined) throw new Error("missing");
    await tx.accounts.update({ where: { id: "ada" }, set: { balance: from.balance - 40 } });
    await tx.accounts.update({ where: { id: "grace" }, set: { balance: to.balance + 40 } });
  });
  const rows = await db.accounts.find({ orderBy: { id: "asc" }, limit: 2 });
  console.log(JSON.stringify(rows.map((row) => ({ id: row.id, balance: row.balance }))));
}

committed
[{"id":"ada","balance":60},{"id":"grace","balance":90}]

A rollback

The callback writes, then throws. The write is gone. afterCommit does not run.

import { connect } from "okmodel/pg/postgresjs";
import { schema, table, t } from "okmodel/pg";

const accounts = table("accounts", {
  id: t.text().primaryKey(),
  balance: t.integer(),
});
export const app = schema({ tables: [accounts] });

export async function rollback(url: string): Promise<void> {
  await using db = connect(url, { schema: app });
  await db.accounts.insert([
    { id: "ada", balance: 100 },
    { id: "grace", balance: 50 },
  ]);
  let ran = false;
  try {
    await db.tx(async (tx) => {
      tx.afterCommit(() => {
        ran = true;
      });
      await tx.accounts.update({ where: { id: "ada" }, set: { balance: 0 } }); 
      throw new Error("stop");
    });
  } catch (error) {
    console.log(error instanceof Error ? error.message : "error");
  }
  console.log(String(ran));
  const rows = await db.accounts.find({ orderBy: { id: "asc" }, limit: 2 });
  console.log(JSON.stringify(rows.map((row) => ({ id: row.id, balance: row.balance }))));
}

stop
false
[{"id":"ada","balance":100},{"id":"grace","balance":50}]

A serialization failure

Two serializable transactions read the same balances. The second writes after the first commits, so the second fails once and runs again. retry: 3 allows that extra run. The first line is how many times the callbacks ran.

import { connect } from "okmodel/pg/postgresjs";
import { schema, table, t } from "okmodel/pg";

const accounts = table("accounts", {
  id: t.text().primaryKey(),
  balance: t.integer(),
});
export const app = schema({ tables: [accounts] });

export async function retry(url: string): Promise<void> {
  await using db = connect(url, { schema: app });
  await db.accounts.insert([
    { id: "a", balance: 100 },
    { id: "b", balance: 100 },
  ]);
  const attempts = { a: 0, b: 0 };
  let waiting = 0;
  let open = (): void => undefined;
  const both = new Promise<void>((resolve) => {
    open = resolve;
  });
  let releaseB = (): void => undefined;
  const aCommitted = new Promise<void>((resolve) => {
    releaseB = resolve;
  });
  const run = (me: "a" | "b") =>
    db.tx({ isolation: "serializable", retry: 3 }, async (tx) => {
      attempts[me] += 1;
      const rows = await tx.accounts.find({ limit: 10 });
      const total = rows.reduce((sum, row) => sum + row.balance, 0);
      if (attempts[me] === 1) {
        waiting += 1;
        if (waiting === 2) open();
        await both;
      }
      if (me === "b" && attempts.b === 1) await aCommitted;
      await tx.accounts.update({ where: { id: me }, set: { balance: total } });
    });
  const a = run("a").finally(() => {
    releaseB();
  });
  await Promise.all([a, run("b")]);
  console.log(String(attempts.a + attempts.b));
  const rows = await db.accounts.find({ orderBy: { id: "asc" }, limit: 2 });
  console.log(JSON.stringify(rows.map((row) => ({ id: row.id, balance: row.balance }))));
}

3
[{"id":"a","balance":200},{"id":"b","balance":300}]