feat(database): put transaction settings in BEGIN through SqlClient.BeginTransaction

.changeset/begin-transaction-settings.md

@@ -0,0 +1,5 @@
+---
+"@maple-dev/effect-orm": minor
+---
+
+`Db.transaction({ isolationLevel, accessMode, deferrable })` on Postgres puts the settings in the statement that starts the transaction (`BEGIN ISOLATION LEVEL SERIALIZABLE, READ ONLY`) through Effect's `SqlClient.BeginTransaction`, one round trip less than the `SET TRANSACTION` it sent before. A dialect gives the BEGIN statement with the new optional `DialectTransactions.beginTransaction`; `postgresDialect` has it, and a dialect without it still sends `SET TRANSACTION`. SQLite keeps its client's own BEGIN. `observe` sees the BEGIN statement, and a failed BEGIN reports it as `DatabaseError.sql`.

design/transactions.md

@@ -68,6 +68,11 @@ client (`beginTransaction`, default `"BEGIN"`; ClickHouse passes `"BEGIN TRANSAC
 a `SET TRANSACTION ...` statement as the first statement inside the transaction, which
 Postgres accepts until the first query of the transaction.
 
+effect 4.0.6 adds `SqlClient.BeginTransaction` (Effect-TS/effect#9576): a
+`Context.Reference<string | undefined>` that, when set, replaces the client's BEGIN for a
+top-level transaction (sent unprepared; savepoints ignore it). `transaction` uses it for a
+dialect with `beginTransaction`; see section 9.
+
 **Tracing.** A `sql.transaction` span with `db.transaction.commit`,
 `db.transaction.savepoint` and `db.transaction.rollback` events; statements inside are
 children of it.
@@ -448,14 +453,18 @@ export interface Dialect extends SqlSyntax {
 ```
 
 - `postgresDialect`: `full`, savepoints, the three levels, access modes, deferrable,
-  `SET TRANSACTION ...`.
+  `BEGIN ...` through `beginTransaction` (`SET TRANSACTION ...` for a dialect without it).
 - `clickhouseDialect`: `none`. A later `"experimental-session"` value would be added only with
   the session-bound executor from section 3.
 - MySQL later: `full`, savepoints, four levels; it needs `SET TRANSACTION` **before** BEGIN,
   so the hook may need a `placement: "before-begin" | "after-begin"`, which Effect's
   `withTransaction` cannot do for a pooled client. That is the point to revisit, not now.
 - SQLite later: `full`, savepoints, no isolation levels (it is serializable), access mode via
   `BEGIN DEFERRED/IMMEDIATE` rather than options.
+- With `SqlClient.BeginTransaction` (section 9) a dialect can also give `beginTransaction`,
+  the whole BEGIN statement. That is the hook for SQLite's `BEGIN IMMEDIATE` and MySQL's
+  `START TRANSACTION READ ONLY`. MySQL's isolation level still needs a `SET TRANSACTION`
+  before BEGIN, which neither hook can do on a pooled client.
 
 Capability is runtime, checked before any statement. A type-level split (no `transaction`
 method on a ClickHouse database) was considered and deferred: the `Dialect` value is not a
@@ -650,6 +659,15 @@ What landed differently from the sections above:
 - **Not yet done from phase 2:** the over-the-wire `@effect/sql-pg` suite (two connections,
   real serialization failures), and filing the upstream Effect issue for orDie'd COMMIT and
   ROLLBACK. The PGlite suite simulates contention with `RAISE ... USING ERRCODE`.
+- **Settings in BEGIN (`SqlClient.BeginTransaction`, effect 4.0.6).** `DialectTransactions.beginTransaction`
+  (optional) writes the BEGIN that carries the settings; Postgres writes
+  `BEGIN ISOLATION LEVEL ..., READ ONLY, DEFERRABLE`. When the dialect has the hook,
+  `transaction` provides the statement around `withTransaction` and sends no
+  `SET TRANSACTION`; a dialect without it keeps `SET TRANSACTION`. SQLite has neither: its
+  client keeps its own `BEGIN IMMEDIATE`, and serializable is the only level it accepts. The
+  reference is cleared inside the body: Effect applies it to every top-level transaction
+  started while it is provided, so a transaction the body starts on a second client would take
+  it too. `observe` sees the BEGIN statement, so a statement collector still sees the settings.
 
 ## 10. `afterCommit` (issue #73)
 

docs/database.md

@@ -144,7 +144,7 @@ For SQL inside a builder query rather than a whole statement, use
 | --- | --- |
 | `dialect` | `postgresDialect` or `clickhouseDialect` |
 | `command` | Wraps `execute`. Pass ClickHouse's `asCommand`: DDL has no JSON result |
-| `observe` | Called with every statement before it runs, including the `SET TRANSACTION` a transaction's settings become |
+| `observe` | Called with every statement before it runs, including the `BEGIN ...` or `SET TRANSACTION` a transaction's settings become |
 
 `run` refuses a query compiled for another dialect (`CompiledQuery.dialect`), as a defect: a
 query compiled with the `/clickhouse` entry's `compile` would run on Postgres with the wrong
@@ -247,7 +247,13 @@ transaction that has already ended (a fiber forked inside it and not joined).
 | `deferrable` | With `serializable` and `read only`, wait for a safe snapshot |
 | `retry` | `"contention"`, or `{ times, schedule }` |
 
-Settings become one `SET TRANSACTION ...` statement, run first. A nested transaction cannot
+On Postgres, settings go into the statement that starts the transaction,
+`BEGIN ISOLATION LEVEL SERIALIZABLE, READ ONLY`, through Effect's `SqlClient.BeginTransaction`,
+which saves the round trip of a separate `SET TRANSACTION`. On SQLite, `serializable` is the
+only level, so there is nothing to send and the client keeps its own BEGIN (`BEGIN IMMEDIATE`
+for `@effect/sql-sqlite-node`). Inside the body,
+`BeginTransaction` is cleared, so a transaction the body starts on another client keeps that
+client's own BEGIN. A nested transaction cannot
 set any of them, or `retry`, and fails with `TransactionOptionsRejected`: Postgres fixes the
 isolation level at the first statement, and a retry has to start the outermost transaction
 again. `read uncommitted` is not offered because Postgres runs it as `read committed`.

docs/reference.md

@@ -89,7 +89,9 @@ to its own dialect; `options.dialect` overrides it.
 `Dialect`, `DialectClauses` and `ParamStyle` describe a database: how identifiers and literals
 are written, how params reach the server, and which clauses exist. Pass one as
 `options.dialect`. `DialectTransactions` (with `IsolationLevel` and `TransactionSettings`) says
-which transactions the database supports; see [Database](./database.md). See [Params and compilation](./params-and-compilation.md#dialects) and
+which transactions the database supports, and how settings are written: `beginTransaction`
+(optional) for a BEGIN that carries them, `setTransaction` for a statement run after BEGIN;
+see [Database](./database.md#settings). See [Params and compilation](./params-and-compilation.md#dialects) and
 [Postgres](./postgres.md).
 
 ### Params

src/ch/dialect.ts

@@ -128,6 +128,15 @@ export interface DialectTransactions {
 	 * transaction, or `undefined` when there is nothing to apply.
 	 */
 	readonly setTransaction: (settings: TransactionSettings) => string | undefined
+	/**
+	 * The statement that starts a transaction with `settings` applied, such as
+	 * `BEGIN ISOLATION LEVEL SERIALIZABLE`, or `undefined` to keep the client's
+	 * BEGIN. Given to Effect's `SqlClient.BeginTransaction` in place of
+	 * `setTransaction`. A dialect without it gets `setTransaction`. A client
+	 * that does not start transactions with SQL ignores it, so leave it out
+	 * where settings must not be dropped silently.
+	 */
+	readonly beginTransaction?: (settings: TransactionSettings) => string | undefined
 }
 
 /** No transactions. What a dialect without a `transactions` entry gets. */

src/database/after-commit.test.ts

@@ -78,6 +78,33 @@ layer(Live, { excludeTestServices: true })("afterCommit on PGlite", (it) => {
 		}),
 	)
 
+	// The settings go in through `SqlClient.BeginTransaction`; a hook runs after
+	// COMMIT and must not start its own transaction with them.
+	it.effect("runs after a transaction with settings, and a transaction it starts keeps the default BEGIN", () =>
+		Effect.gen(function* () {
+			const table = yield* freshTable
+			const seen = yield* Ref.make<ReadonlyArray<unknown>>([])
+			yield* Db.transaction(
+				Effect.gen(function* () {
+					yield* insert(table, 1)
+					yield* Db.afterCommit(
+						Effect.gen(function* () {
+							const inHook = yield* Effect.service(SqlClient.BeginTransaction)
+							const rows = yield* Db.transaction(
+								Db.query(
+									Db.sql`SELECT current_setting('transaction_isolation') AS isolation, current_setting('transaction_read_only') AS read_only`,
+								),
+							)
+							yield* Ref.set(seen, [inHook, rows[0]])
+						}).pipe(Effect.orDie),
+					)
+				}),
+				{ isolationLevel: "repeatable read" },
+			)
+			expect(yield* Ref.get(seen)).toEqual([undefined, { isolation: "read committed", read_only: "off" }])
+		}),
+	)
+
 	it.effect("runs in queue order; a rolled-back savepoint drops its hooks, a kept one hands them up", () =>
 		Effect.gen(function* () {
 			const ran = yield* Ref.make<ReadonlyArray<string>>([])

src/database/database.test.ts

@@ -44,6 +44,17 @@ const ids = (table: string) =>
 const raiseSqlState = (code: "40001" | "40P01") =>
 	Db.execute({ sql: `DO $$ BEGIN RAISE EXCEPTION 'forced' USING ERRCODE = '${code}'; END $$` })
 
+// Records the `SqlClient.BeginTransaction` each top-level transaction starts
+// with, then runs the real transaction.
+const recordingBegin = (sql: SqlClient.SqlClient, begins: Array<string | undefined>): SqlClient.SqlClient =>
+	Object.assign(Object.create(sql) as SqlClient.SqlClient, {
+		withTransaction: <A, E, R>(effect: Effect.Effect<A, E, R>) =>
+			Effect.flatMap(Effect.service(SqlClient.BeginTransaction), (begin) => {
+				begins.push(begin)
+				return sql.withTransaction(effect)
+			}),
+	})
+
 class Domain extends Error {
 	readonly _tag = "Domain"
 }
@@ -214,7 +225,7 @@ layer(Live, { excludeTestServices: true })("Database on PGlite", (it) => {
 		}),
 	)
 
-	it.effect("applies settings as the first statement of the transaction", () =>
+	it.effect("applies settings when the transaction starts", () =>
 		Effect.gen(function* () {
 			statements.length = 0
 			const settings = yield* Db.transaction(
@@ -227,12 +238,51 @@ layer(Live, { excludeTestServices: true })("Database on PGlite", (it) => {
 				}),
 				{ isolationLevel: "serializable", accessMode: "read only", deferrable: true },
 			)
-			expect(statements[0]).toBe("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE")
+			// The settings travel in BEGIN; the current_setting values show PGlite ran it.
+			expect(statements[0]).toBe("BEGIN ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE")
+			expect(statements).not.toContain("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE")
 			expect(settings.row).toEqual({ isolation: "serializable", read_only: "on" })
 			expect(settings.info).toEqual({ depth: 0, isolationLevel: "serializable", accessMode: "read only" })
 		}),
 	)
 
+	it.effect("with Effect's BeginTransaction, settings go into the BEGIN statement and no SET TRANSACTION is sent", () =>
+		Effect.gen(function* () {
+			const begins: Array<string | undefined> = []
+			const sent: Array<string> = []
+			const db = Db.fromSqlClient(recordingBegin(yield* SqlClient.SqlClient, begins), {
+				dialect: postgresDialect,
+				observe: (statement) => Effect.sync(() => void sent.push(statement.sql)),
+			})
+			const inBody = yield* db.transaction(
+				Effect.andThen(db.execute(Db.sql`SELECT 1`), Effect.service(SqlClient.BeginTransaction)),
+				{ isolationLevel: "serializable", accessMode: "read only", deferrable: true },
+			)
+			yield* db.transaction(db.execute(Db.sql`SELECT 2`))
+			expect(begins).toEqual(["BEGIN ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE", undefined])
+			expect(sent).toEqual(["BEGIN ISOLATION LEVEL SERIALIZABLE, READ ONLY, DEFERRABLE", "SELECT 1", "SELECT 2"])
+			// Not seen inside the body, so a transaction it starts on another client keeps that client's BEGIN.
+			expect(inBody).toBeUndefined()
+		}),
+	)
+
+	it.effect("a dialect without beginTransaction sends its settings as a SET TRANSACTION run first", () =>
+		Effect.gen(function* () {
+			const sent: Array<string> = []
+			const { beginTransaction: _, ...transactions } = postgresDialect.transactions!
+			const db = Db.fromSqlClient(yield* SqlClient.SqlClient, {
+				dialect: { ...postgresDialect, transactions },
+				observe: (statement) => Effect.sync(() => void sent.push(statement.sql)),
+			})
+			const row = yield* db.transaction(
+				db.query(Db.sql`SELECT current_setting('transaction_isolation') AS isolation`),
+				{ isolationLevel: "repeatable read" },
+			)
+			expect(sent[0]).toBe("SET TRANSACTION ISOLATION LEVEL REPEATABLE READ")
+			expect(row).toEqual([{ isolation: "repeatable read" }])
+		}),
+	)
+
 	it.effect("a write under read only fails as DatabaseError with its SQLSTATE", () =>
 		Effect.gen(function* () {
 			const table = yield* freshTable

src/database/database.ts

@@ -149,9 +149,9 @@ export interface FromSqlClientOptions {
 	 */
 	readonly command?: <A, E, R>(effect: Effect.Effect<A, E, R>) => Effect.Effect<A, E, R>
 	/**
-	 * Called with every statement before it runs, including the `SET
-	 * TRANSACTION` a transaction's settings become. For collecting statements
-	 * into one span or log line per request.
+	 * Called with every statement before it runs, including the `BEGIN ...`
+	 * (effect 4.0.6 and later) or `SET TRANSACTION` a transaction's settings
+	 * become. For collecting statements into one span or log line per request.
 	 */
 	readonly observe?: (statement: Statement) => Effect.Effect<void>
 	/**
@@ -384,14 +384,24 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp
 				depth === 0 || Option.isNone(outer)
 					? { depth, isolationLevel: settings.isolationLevel, accessMode: settings.accessMode }
 					: { ...outer.value, depth }
-			const setTransaction = depth === 0 ? capabilities.setTransaction(settings) : undefined
+			// Settings travel in BEGIN when the dialect can write one, else as a
+			// `SET TRANSACTION` run first. Top level only: a savepoint takes no
+			// settings. Without settings the client keeps its own BEGIN (SQLite's
+			// `BEGIN IMMEDIATE`, for one).
+			const beginHook = capabilities.beginTransaction
+			const begin = depth === 0 && beginHook !== undefined ? beginHook(settings) : undefined
+			const setTransaction = depth === 0 && beginHook === undefined ? capabilities.setTransaction(settings) : undefined
 			const state: OpenState = { open: true, client: sql }
 			let bodyExit: Exit.Exit<unknown, unknown> | undefined
 
 			const inner = Effect.gen(function* () {
 				if (setTransaction !== undefined) yield* execute({ sql: setTransaction })
 				return yield* body
 			}).pipe(
+				// The reference applies to every top-level transaction started while it
+				// is provided; clear it for the body, so a transaction on another client
+				// keeps that client's BEGIN.
+				(effect) => (begin === undefined ? effect : Effect.provideService(effect, SqlClient.BeginTransaction, undefined)),
 				Effect.provideService(Transaction, info),
 				Effect.provideService(TransactionOpen, state),
 				Effect.onExit((exit) =>
@@ -404,9 +414,16 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp
 				),
 			)
 
-			const exit = yield* Effect.exit(sql.withTransaction(inner).pipe(Effect.ensuring(committed(state))))
+			const transaction =
+				begin === undefined
+					? sql.withTransaction(inner)
+					: Effect.andThen(
+							observe({ sql: begin }),
+							sql.withTransaction(inner).pipe(Effect.provideService(SqlClient.BeginTransaction, begin)),
+						)
+			const exit = yield* Effect.exit(transaction.pipe(Effect.ensuring(committed(state))))
 			if (Exit.isSuccess(exit)) return exit.value
-			return yield* restoreCause<E>(exit.cause, bodyExit, depth)
+			return yield* restoreCause<E>(exit.cause, bodyExit, depth, begin)
 		}).pipe(
 			Effect.withSpan("effect_orm.transaction", {
 				// Semconv defines no transaction attributes, and `db.*` is reserved for it.
@@ -430,12 +447,14 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp
  * Effect fails with the `SqlError` of a failed COMMIT (or dies, when cleanup
  * after it failed too), and dies on a failed ROLLBACK, which replaces the body's
  * own exit. Body defects were carried through as `BodyDefect`, so any other
- * defect that is a `SqlError` came from transaction control.
+ * defect that is a `SqlError` came from transaction control. `begin` is the
+ * statement that started the transaction, when it was not the client's BEGIN.
  */
 const restoreCause = <E>(
 	cause: Cause.Cause<unknown>,
 	bodyExit: Exit.Exit<unknown, unknown> | undefined,
 	depth: number,
+	begin: string | undefined,
 ): Effect.Effect<never, E | DatabaseError | TransactionCommitFailed | TransactionRollbackFailed> => {
 	const failure = Cause.findError(cause)
 	if (failure._tag === "Success" && failure.success instanceof BodyDefect) {
@@ -464,7 +483,7 @@ const restoreCause = <E>(
 	if (defect._tag === "Success" && isSqlError(defect.success)) {
 		const error = defect.success
 		// The body never ran: BEGIN or SAVEPOINT died, not ROLLBACK.
-		if (bodyExit === undefined) return Effect.fail(toDatabaseError(depth === 0 ? "BEGIN" : "SAVEPOINT")(error))
+		if (bodyExit === undefined) return Effect.fail(toDatabaseError(depth === 0 ? (begin ?? "BEGIN") : "SAVEPOINT")(error))
 		// An interrupted transaction whose ROLLBACK died stays interrupted.
 		if (Cause.hasInterrupts(bodyExit.cause)) return Effect.failCause(bodyExit.cause as Cause.Cause<E>)
 		return Effect.fail(
@@ -477,7 +496,7 @@ const restoreCause = <E>(
 	}
 	// BEGIN or SAVEPOINT failed before the body ran.
 	if (bodyExit === undefined && failure._tag === "Success" && isSqlError(failure.success)) {
-		return Effect.fail(toDatabaseError(depth === 0 ? "BEGIN" : "SAVEPOINT")(failure.success))
+		return Effect.fail(toDatabaseError(depth === 0 ? (begin ?? "BEGIN") : "SAVEPOINT")(failure.success))
 	}
 	return Effect.failCause(cause as Cause.Cause<E>)
 }

src/pg/dialect.ts

@@ -1,7 +1,7 @@
 // The Postgres dialect.
 
 import { DateTime, Schema } from "effect"
-import type { Dialect } from "../ch/dialect"
+import type { Dialect, TransactionSettings } from "../ch/dialect"
 import { QueryBuilderError } from "../ch/errors"
 import { PARAM_MARKER_PREFIX } from "../ch/param"
 import { PgTimestampLiteral, timestampLiteral } from "./types"
@@ -77,6 +77,16 @@ const placeholderCasts: Readonly<Record<string, string>> = {
 	dateTimeSeconds: "timestamptz",
 }
 
+// Postgres takes the same mode list after `SET TRANSACTION` and after `BEGIN`.
+const transactionModes = (settings: TransactionSettings): string | undefined => {
+	const modes = [
+		settings.isolationLevel === undefined ? undefined : `ISOLATION LEVEL ${settings.isolationLevel.toUpperCase()}`,
+		settings.accessMode?.toUpperCase(),
+		settings.deferrable === undefined ? undefined : settings.deferrable ? "DEFERRABLE" : "NOT DEFERRABLE",
+	].filter((mode) => mode !== undefined)
+	return modes.length === 0 ? undefined : modes.join(", ")
+}
+
 /**
  * Postgres: double-quoted identifiers, standard string literals, and params
  * bound to `$1`, `$2`, … and returned in `CompiledQuery.parameters`.
@@ -120,12 +130,12 @@ export const postgresDialect: Dialect = {
 		accessModes: true,
 		deferrable: true,
 		setTransaction: (settings) => {
-			const modes = [
-				settings.isolationLevel === undefined ? undefined : `ISOLATION LEVEL ${settings.isolationLevel.toUpperCase()}`,
-				settings.accessMode?.toUpperCase(),
-				settings.deferrable === undefined ? undefined : settings.deferrable ? "DEFERRABLE" : "NOT DEFERRABLE",
-			].filter((mode) => mode !== undefined)
-			return modes.length === 0 ? undefined : `SET TRANSACTION ${modes.join(", ")}`
+			const modes = transactionModes(settings)
+			return modes === undefined ? undefined : `SET TRANSACTION ${modes}`
+		},
+		beginTransaction: (settings) => {
+			const modes = transactionModes(settings)
+			return modes === undefined ? undefined : `BEGIN ${modes}`
 		},
 	},
 }