# database.transactions > How ctx.store behaves inside mutations, workflows, and explicit transaction blocks. --- ## Transactions _How ctx.store behaves inside mutations, workflows, and explicit transaction blocks._ Every mutation executor runs inside an implicit Postgres transaction. Workflows manage transactions per step. You rarely call `BEGIN`/`COMMIT` directly — but when you do, the API is clean. ## The default: one mutation = one transaction ```tsx // apps/api/mutations/orders.create.mutation.ts export default async (input, ctx) => { const order = await ctx.store.insert('orders', { /* … */ }) for (const line of input.lines) { await ctx.store.insert('orderLines', { orderId: order.id, ...line }) } return { id: order.id } } ``` The framework wraps this in `BEGIN; … COMMIT;`. If the executor throws (a typed error or any unhandled exception), the transaction rolls back — no half-inserted orders. CDC events fire only AFTER commit. Subscribers don't see uncommitted writes. ## Explicit transactions When you need a transaction across multiple top-level operations (rare in mutations, which are already transactional; common in setup scripts + workflow steps), use `ctx.store.transactional(work)`: ```ts await ctx.store.transactional(async (tx) => { await tx.insert('users', { /* … */ }) await tx.insert('teams', { /* … */ }) await tx.insert('memberships', { /* … */ }) // Throwing here rolls back all three. }) ``` `tx` is a `ctx.store` clone scoped to one transaction. Reads via `tx.query(...)` / `tx.select(...)` inside see the in-flight write set (read-your-own-writes within the transaction). ChangeEvents queue inside the transaction and emit ONLY after a successful commit, so subscribers never see a partial-update flicker. `transactional()` **must not be nested** — a nested invocation throws to surface a design mistake. Use a single top-level `transactional()` per unit of work. ## Typed errors survive a transaction A `Data.TaggedError` thrown inside a transaction reaches the caller — and the browser client — **as itself**: `_tag`, payload and prototype intact, so a mutation's declared `error:` union still matches. ```ts import { Data } from 'effect' class NoteNotFound extends Data.TaggedError('NoteNotFound')<{ noteId: string }> {} // apps/api/mutations/notes.rename.mutation.ts — auto-transactional export default async (input, ctx) => { const note = await ctx.store.query({ table: 'notes', /* … */ }) if (!note) throw new NoteNotFound({ noteId: input.id }) // arrives typed on the client // … } ``` On the client: ```ts const res = await client.notes.rename({ id, title }) if (res.error?._tag === 'NoteNotFound') { /* this branch fires */ } ``` This holds on **every** dialect and on **both** tenancy topologies — shared-schema and namespace isolation — because all four dialect stores and both postgres entry points run through one shared transaction bracket. Never match on `error.message` to identify a transaction failure; the `_tag` is the contract. ## Automatic retry on transient conflicts A transaction that fails with a transient contention error is retried automatically: exponential backoff from 10 ms, up to 3 retries (4 attempts total). What counts as transient is per-dialect — postgres `serialization_failure` (40001) and `deadlock_detected` (40P01), mysql/mariadb deadlock and lock-wait timeout, mssql deadlock victim (1205), sqlite `SQLITE_BUSY`/`SQLITE_LOCKED`, and Turso's MVCC write-write conflict. Two properties worth knowing: - **A conflict raised by the COMMIT itself is retried too.** Under `SERIALIZABLE`, the engine can only detect some conflicts at commit time — those are caught and replayed like any other. - **Each attempt gets a fresh transaction and a fresh event buffer.** A retried attempt's queued ChangeEvents are discarded with it, so subscribers see exactly one event set: the winning attempt's. Because the body can run more than once, keep `transactional()` bodies **idempotent** — no counters incremented in JS, no external calls (see the anti-patterns below). ## Optimistic concurrency The fluent update builder carries an `.expectVersion(n)` guard: the update only matches rows whose `version` equals `n`, and throws `OptimisticLockError` when none match (the row was concurrently changed). Add a `version` column to the table to use it. ```ts await ctx.store.update('notes').where('id', id).expectVersion(3).set({ title: 'x' }) ``` ## Workflows + transactions Workflows are step-machines, not transactions — a workflow spans many transactions across many processes. **Don't** wrap a whole workflow in a transaction. Instead, scope each `yield* step()` to its own transaction: ```tsx import { Effect } from 'effect' export default ({ orderId }) => Effect.gen(function* () { yield* recordPaymentAttempt(orderId) // tx 1 — commits before the next step starts yield* chargeStripe(orderId) // external — no transaction yield* markOrderPaid(orderId) // tx 2 }) ``` If the worker dies between `chargeStripe` + `markOrderPaid`, the next worker picks up at `markOrderPaid` — the durable workflow log knows tx 1 committed already. ## CDC + reactivity Mutation writes trigger Postgres logical replication events. The runtime reads from a replication slot, decodes each WAL message, and queries invalidations to every subscription whose tracked read set touches the changed row. This means: - **Writes from outside the framework** (psql, dump-restore, external workers) ARE picked up. Subscribers see them like any other write. - **Writes inside `pg_dump` / `pg_restore`** bypass logical replication and DON'T invalidate. Always `voltro migrate` after a restore. - **Uncommitted writes don't invalidate.** Subscribers see the post-commit state, never a torn read. ## Connection pooling Voltro runs on `@effect/sql`'s per-dialect driver, which manages its own connection pool. Connection details flow from the standard env vars (`PG_HOST` / `PG_PORT` / `PG_USER` / `PG_PASSWORD` / `PG_DATABASE` for postgres, the `MYSQL_*` / `MARIADB_*` / `MSSQL_*` / `DB_URL` analogues for the others). For high-concurrency setups (1k+ rps), front Postgres with PgBouncer in transaction mode. ## Anti-patterns - **Long transactions.** Keep `transactional()` bodies short — the longer the BEGIN…COMMIT window, the more contention. Do external I/O (HTTP, AI calls) OUTSIDE the transaction, in an action or a workflow step. - **Nesting `transactional()`.** It throws. Build one top-level transaction per unit of work. - **Cross-transaction state in workflows.** If a workflow needs "transactional" semantics across steps, it doesn't — it needs a saga (compensating actions on failure). - **Reading from `ctx.store` after throwing.** The transaction is rolled back — your write is gone. Plan for compensation in the caller. --- ## Bulk write operations (insertMany / updateMany / upsert / insertIgnore) _Insert many rows in one statement, update many rows in one statement, upsert with ON CONFLICT semantics, insert-or-ignore for idempotent writes._ Four write APIs that go beyond single-row insert/update: - **`insertMany(table, rows)`** — one-statement multi-row insert - **`updateMany(table, patch, { where })`** — one-statement bulk update - **`upsert(table, row, options)`** — insert OR update on conflict - **`insertIgnore(table, row, options)`** — insert OR keep existing All four live on `ctx.store` alongside `insert` / `update` / `delete`. ## `insertMany` — one-statement multi-row insert ```ts const inserted = await ctx.store.insertMany('todos', [ { title: 'a', done: false }, { title: 'b', done: false }, { title: 'c', done: false }, ]) // inserted: the rows as persisted, in input order (ids auto-injected) ``` One multi-row `INSERT ... VALUES (...),(...)` per dialect — a single round-trip regardless of row count, instead of N `insert()` calls. The auto-stamping middleware (id injection, audit/tenant columns, `.computed()` columns, `.validate()`) runs per row exactly as `insert()` does. An empty array is a no-op that returns `[]`. ### Reactivity `insertMany` emits ONE insert ChangeEvent per row, so reactive subscribers see each new row's delta — identical to N single inserts. Inside a `transactional()` the per-row events queue until commit and drop on a throw. ### Large arrays are chunked for you — and stay all-or-nothing Every engine caps what ONE statement may carry, and the caps are far apart: | Dialect | Bind parameters per statement | Rows per `VALUES` | | --- | --- | --- | | postgres | 65 535 | — | | mysql / mariadb | 65 535 | — | | **mssql** | **2 098** | **1 000** | | sqlite / turso | 32 766 | — | `INSERT … VALUES` binds one parameter per **column per row**, so the row limit is `floor(cap / columns)` — a 12-column table caps at 5 461 rows on postgres and at **174** on mssql. Past that the driver refuses with its own error about a limit you never chose. `insertMany` splits the array for you at that boundary. Two properties are guaranteed: - **A fitting array is still ONE statement.** Nothing changes for the normal case — no extra round-trips, no behaviour difference. - **A chunked insert is still all-or-nothing.** When the call is not already inside a `transactional()`, the chunks run in one transaction the framework opens, so a failure in the last chunk rolls back the earlier ones. Without that, chunking would quietly add partial-success-on-failure to a call that never had it. There is nothing to configure. If a SINGLE row is wider than the cap (a 3 000-column table on mssql) the engine's own error is what you get — a row cannot be split. ## `upsert` — insert or update on conflict The 90% case: "make this row exist with these values; if it already exists, update it": ```ts await ctx.store.upsert('users', { email: input.email, displayName: input.displayName, }, { conflictColumns: ['email'], }) ``` `conflictColumns` declares which columns identify "the same row". Typically a single column with a UNIQUE constraint (`['email']`) or a composite UNIQUE (`['orgId', 'slug']`). ### Three update strategies **Default — overwrite every input column:** ```ts await ctx.store.upsert('users', input, { conflictColumns: ['email'], // update: omitted → every non-id, non-conflict column from `input` }) ``` **Whitelist — only certain columns:** ```ts await ctx.store.upsert('users', input, { conflictColumns: ['email'], update: ['displayName'], // overwrite ONLY displayName on conflict }) ``` **Function — compute patch from existing row:** ```ts await ctx.store.upsert('users', input, { conflictColumns: ['email'], update: (existing) => ({ visits: (existing.visits ?? 0) + 1 }), }) // Increment counter pattern — read then patch ``` ### Composite conflict keys ```ts await ctx.store.upsert('orgSlugs', { orgId: 'o1', slug: 'dashboard', active: true, }, { conflictColumns: ['orgId', 'slug'], }) ``` Requires a composite UNIQUE constraint on the table — declare it via `.unique([cols])` in the schema (see [Indexes](/docs/database/indexes#composite-unique)). ## `insertIgnore` — keep existing on conflict For idempotent write patterns where you want to ensure a row exists but DON'T want to overwrite it: ```ts await ctx.store.insertIgnore('audit_dedup', { eventId, ts: new Date(), }, { conflictColumns: ['eventId'], }) // First call: row inserted, returned // Second call (same eventId): existing row returned, no write happens ``` Common uses: - Dedup events by external ID (Stripe webhooks, Shopify orders) - Idempotent "ensure tenant exists" patterns - Audit logs where re-running a sync shouldn't double-write Returns the FINAL row in both cases (newly-inserted or pre-existing). ### One conflict target, and only conflicts `conflictColumns` names ONE constraint. A duplicate on a *different* unique index is not something `insertIgnore` can resolve — it cannot know which existing row you meant — so it throws, naming the constraint that actually fired. It also throws when the insert was **rejected** rather than skipped. This matters most on MariaDB, where the statement lowers to `INSERT IGNORE`: that downgrades *every* error to a warning — foreign key, NOT NULL, CHECK, truncation — not just the unique violation the API models. So "no row was inserted" does not imply "a conflict happened", and reporting one as the other would turn a rejected write into a silent no-op: the row is not there and the caller is told it already was. ``` MysqlStore.insertIgnore: the insert into 'docs' was REJECTED, not skipped as a conflict. INSERT IGNORE downgrades every error to a warning, and the warning was: [1452] Cannot add or update a child row: a foreign key constraint fails … Nothing was written and nothing conflicted — fix the cause above. ``` The real cause comes from `SHOW WARNINGS` on the same connection, which is only attributable inside a transaction — so outside one the message says the constraint is unknown rather than guessing at it. Framework mutations are auto-transactional, so the common path has the cause. ## `updateMany` — one-statement bulk update ```ts import { inSubquery, eq, queryFor } from '@voltro/database' // Hide every post by a banned user const count = await ctx.store.updateMany('posts', { hidden: true }, { where: inSubquery('userId', queryFor(database.users).where(eq('banned', true)).select('id'), ), }) // count: number of rows actually updated ``` The `where` predicate is a regular [Predicate](/docs/database/query-builder#predicates) AST — same shape `.where()` uses. Sub-queries via `inSubquery` / `exists` are supported. ### Typed: `updateManyRow` `updateMany` takes a string table name and an untyped patch, so a misspelled column or a wrongly-typed value is only found by the database — or not at all, if the column happens to exist. `updateManyRow` takes the TABLE OBJECT instead and checks the patch against the row type: ```ts import { updateManyRow, eq } from '@voltro/database' await updateManyRow(ctx.store, posts, { hidden: true }, { where: eq('userId', bannedId), }) await updateManyRow(ctx.store, posts, { hiddne: true }, { where: … }) // ^^^^^^^ compile error: not a column ``` It is worth using rather than the string form, and the evidence is concrete: one app migrating 29 `store.upsert` call sites to the typed `upsertRow` got 15 `tsc` errors across 8 distinct defects that no test had caught — including seven per-user mutations with no authentication check at all, which wrote `ctx.request.subject.id` (typed `string | null`) into a NOT NULL column, so an anonymous caller reached the database and got a raw statement failure instead of a typed refusal. The subtlest one is the most persuasive: a value spread from a plain object literal widens to `string`, and a column's `.oneOf()` union rejects it even though the value IS one of the members. Neither a reviewer nor a test would plausibly find that; only the row type asks the question. (`as const` fixes it.) ### Reactivity `updateMany` emits ONE ChangeEvent per affected row, so reactive subscribers see deltas just like a row-by-row update path. The event fan-out is preserved on every dialect — via `RETURNING *` (postgres/sqlite), `OUTPUT INSERTED.*` (MSSQL), or a post-image `SELECT` (mysql/mariadb — neither ships `UPDATE … RETURNING`). ## Cross-dialect The framework picks the optimal SQL form per dialect — every operation is O(1) statements regardless of row count: | Dialect | upsert / insertIgnore | updateMany | |---------------|-----------------------------------------------|-----------------------------------------| | postgres | `INSERT ... ON CONFLICT (cols) DO UPDATE/DO NOTHING RETURNING *` | `UPDATE ... WHERE ... RETURNING *` | | sqlite 3.35+ | `ON CONFLICT (cols) DO UPDATE/DO NOTHING` + RETURNING | `UPDATE ... WHERE ... RETURNING *` | | mariadb 10.5+ | native `INSERT ... ON DUPLICATE KEY UPDATE ... RETURNING *` / `INSERT IGNORE ... RETURNING *` | 3 statements (`SELECT` ids → `UPDATE` → `SELECT` post-images) — no `UPDATE … RETURNING` in any MariaDB version | | mysql 8.x | same fallback | 3 statements (`SELECT` ids → `UPDATE` → `SELECT` post-images) | | mssql | same fallback | `UPDATE ... SET ... OUTPUT INSERTED.* WHERE` | The user-facing API is identical across all of them — the per-dialect work is hidden in the store implementations. ## Atomicity All three operations participate in the surrounding `transactional()` wrapper. ChangeEvents queue until commit; a throw rolls them back. For `upsert` on the fallback dialects (mysql/mssql), the SELECT-then-INSERT/UPDATE pair is atomic INSIDE the transaction. Without a surrounding transaction, a concurrent writer could race the SELECT — caller is responsible for wrapping when atomicity matters. ## When NOT to use these - **Loop of single-row writes** — use `updateMany` (for updates) or `insertMany` (for inserts) instead. One round-trip beats N every time. ## See also - [Aggregations](/docs/database/aggregations) — `count()` / `aggregate()` for read-side bulk reads - [Sub-queries](/docs/database/sub-queries) — `inSubquery` / `exists` in `updateMany` `where:` clauses - [Composite UNIQUE](/docs/database/indexes#composite-unique) — for the constraint that backs `upsert`'s `conflictColumns: ['a', 'b']`