From 8245897fd46108c866932a55dc7e53fdd86eafdd Mon Sep 17 00:00:00 2001 From: Taras Kornichuk Date: Sun, 25 Oct 2026 17:05:21 +0200 Subject: [PATCH 1/4] feat(database): afterCommit queues work to run after the outermost COMMIT --- src/database.ts | 2 + src/database/after-commit.test.ts | 188 ++++++++++++++++++++++++++++++ src/database/database.test-d.ts | 14 ++- src/database/database.ts | 119 ++++++++++++++++++- src/database/errors.ts | 12 ++ 5 files changed, 329 insertions(+), 6 deletions(-) create mode 100644 src/database/after-commit.test.ts diff --git a/src/database.ts b/src/database.ts index ba66164..865972e 100644 --- a/src/database.ts +++ b/src/database.ts @@ -8,6 +8,7 @@ export { Database, Transaction, + afterCommit, execute, fromSqlClient, isContention, @@ -29,6 +30,7 @@ export { type TransactionOptions, } from "./database/database" export { + AfterCommitUnavailable, DatabaseError, TenantScopeViolation, TransactionClosed, diff --git a/src/database/after-commit.test.ts b/src/database/after-commit.test.ts new file mode 100644 index 0000000..4562f0d --- /dev/null +++ b/src/database/after-commit.test.ts @@ -0,0 +1,188 @@ +// `afterCommit` against a real Effect SqlClient (`@effect/sql-pglite`): work +// that must wait for COMMIT, such as a vendor call, and must never run after a +// rollback. + +import { PgliteClient } from "@effect/sql-pglite" +import { assert, expect, layer } from "@effect/vitest" +import { Cause, Context, Deferred, Effect, ErrorReporter, Exit, Fiber, Layer, Logger, Ref, Schedule } from "effect" +import * as SqlClient from "effect/sql/SqlClient" +import * as CH from "../ch/index" +import * as PG from "../postgres" +import { postgresDialect } from "../pg/dialect" +import * as Db from "../database" + +const Live = Layer.effect( + Db.Database, + Effect.gen(function* () { + return Db.fromSqlClient(yield* SqlClient.SqlClient, { dialect: postgresDialect }) + }), +).pipe(Layer.provideMerge(PgliteClient.layer({ postgresqlconf: "timezone = 'UTC'" }))) + +let tables = 0 +const freshTable = Effect.gen(function* () { + const name = `after_commit_${++tables}` + yield* Db.execute(Db.sql`CREATE TABLE ${Db.sql.identifier(name)} (id int4 PRIMARY KEY)`) + return name +}) +const insert = (table: string, id: number) => Db.execute(Db.sql`INSERT INTO ${Db.sql.identifier(table)} VALUES (${id})`) +const ids = (table: string) => + Db.run(CH.from(CH.table(table, { id: PG.int4 })).select("id").orderBy(["id", "asc"])).pipe( + Effect.map((rows) => rows.map((row) => row.id)), + ) + +class Domain extends Error { + readonly _tag = "Domain" +} + +class Sms extends Context.Service }>()("test/Sms") {} + +const dies = (effect: Effect.Effect) => + Effect.map(Effect.exit(effect), (exit) => { + assert(Exit.isFailure(exit) && Cause.hasDies(exit.cause)) + return Cause.squash(exit.cause) + }) + +layer(Live, { excludeTestServices: true })("afterCommit on PGlite", (it) => { + it.effect("runs after COMMIT, outside the transaction, and never after a rollback", () => + Effect.gen(function* () { + const table = yield* freshTable + const seen = yield* Ref.make>([]) + const record = (label: string) => + Effect.gen(function* () { + const open = yield* Effect.serviceOption(Db.Transaction) + const rows = yield* ids(table) + yield* Ref.update(seen, (all) => [...all, { label, ids: rows, open: open._tag === "Some" }]) + }) + + const value = yield* Db.transaction( + Effect.gen(function* () { + yield* insert(table, 1) + yield* Db.afterCommit(record("sms")) + expect(yield* Ref.get(seen)).toEqual([]) + return "done" + }), + ) + expect(value).toBe("done") + expect(yield* Ref.get(seen)).toEqual([{ label: "sms", ids: [1], open: false }]) + + const error = yield* Effect.flip( + Db.transaction( + insert(table, 2).pipe( + Effect.andThen(Db.afterCommit(record("rolled back"))), + Effect.andThen(Effect.fail(new Domain("no"))), + ), + ), + ) + assert(error instanceof Domain) + expect(yield* Ref.get(seen)).toHaveLength(1) + }), + ) + + 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>([]) + const queue = (label: string) => Db.afterCommit(Ref.update(ran, (all) => [...all, label])) + yield* Db.transaction( + Effect.gen(function* () { + yield* queue("outer 1") + yield* Db.transaction(queue("kept savepoint")) + yield* Db.transaction(queue("dropped savepoint").pipe(Effect.andThen(Effect.fail(new Domain("inner"))))).pipe( + Effect.ignore, + ) + yield* Db.transaction(Db.transaction(queue("two deep"))) + yield* queue("outer 2") + expect(yield* Ref.get(ran)).toEqual([]) + }), + ) + expect(yield* Ref.get(ran)).toEqual(["outer 1", "kept savepoint", "two deep", "outer 2"]) + + // A kept savepoint inside an outer transaction that rolls back runs nothing. + yield* Ref.set(ran, []) + yield* Db.transaction(Db.transaction(queue("inner")).pipe(Effect.andThen(Effect.fail(new Domain("outer"))))).pipe( + Effect.ignore, + ) + expect(yield* Ref.get(ran)).toEqual([]) + }), + ) + + it.effect("a failing hook does not fail the committed transaction: it is reported and logged, and the next hook runs", () => + Effect.gen(function* () { + const table = yield* freshTable + const reported: Array = [] + const logged: Array<{ level: string; message: unknown; failure: unknown }> = [] + const reporter = ErrorReporter.make(({ cause }) => void reported.push(Cause.squash(cause))) + const logger = Logger.make(({ logLevel, message, cause }) => { + logged.push({ level: logLevel, message, failure: Cause.squash(cause) }) + }) + const ran = yield* Ref.make(0) + const vendorDown = new Domain("vendor down") + const bug = new Error("bug in a hook") + + const value = yield* Db.transaction( + Effect.gen(function* () { + yield* insert(table, 1) + yield* Db.afterCommit(Effect.fail(vendorDown)) + yield* Db.afterCommit(Effect.die(bug)) + yield* Db.afterCommit(Ref.update(ran, (n) => n + 1)) + return "committed" + }), + ).pipe(Effect.provide(Layer.merge(ErrorReporter.layer([reporter]), Logger.layer([logger])))) + + expect(value).toBe("committed") + expect(yield* ids(table)).toEqual([1]) + expect(yield* Ref.get(ran)).toBe(1) + expect(reported).toEqual([vendorDown, bug]) + expect(logged.map((entry) => [entry.level, entry.failure])).toEqual([ + ["Error", vendorDown], + ["Error", bug], + ]) + }), + ) + + it.effect("a hook sees the services of the place it was queued, and its R stays required", () => + Effect.gen(function* () { + const sent: Array = [] + const sendSms = (text: string) => Effect.flatMap(Effect.service(Sms), (sms) => Effect.sync(() => void sms.sent.push(text))) + yield* Db.transaction(Db.afterCommit(sendSms("tour booked")).pipe(Effect.provideService(Sms, { sent }))) + expect(sent).toEqual(["tour booked"]) + }), + ) + + it.effect("a retried attempt drops its hooks; only the attempt that commits runs them", () => + Effect.gen(function* () { + const ran = yield* Ref.make>([]) + const attempts = yield* Ref.make(0) + yield* Db.transaction( + Effect.gen(function* () { + const attempt = yield* Ref.updateAndGet(attempts, (n) => n + 1) + yield* Db.afterCommit(Ref.update(ran, (all) => [...all, attempt])) + if (attempt < 3) yield* Db.execute({ sql: "DO $$ BEGIN RAISE EXCEPTION 'forced' USING ERRCODE = '40001'; END $$" }) + }), + { retry: { times: 3, schedule: Schedule.spaced("1 millis") } }, + ) + expect(yield* Ref.get(ran)).toEqual([3]) + }), + ) + + it.effect("dies with AfterCommitUnavailable rather than drop work no transaction will run", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const work = Effect.void + + // The outermost transaction is a plain sql.withTransaction: nothing would run the hook. + const foreign = yield* dies(sql.withTransaction(Db.transaction(Db.afterCommit(work)))) + expect(foreign).toBeInstanceOf(Db.AfterCommitUnavailable) + + // Transaction provided by hand, with no transaction behind it. + const fake = yield* dies(Db.afterCommit(work).pipe(Effect.provideService(Db.Transaction, { depth: 0, isolationLevel: undefined, accessMode: undefined }))) + expect(fake).toBeInstanceOf(Db.AfterCommitUnavailable) + + // A fiber forked in the transaction that queues work after it ended. + const release = yield* Deferred.make() + const straggler = yield* Db.transaction(Deferred.await(release).pipe(Effect.andThen(Db.afterCommit(work)), Effect.forkDetach)) + yield* Deferred.succeed(release, undefined) + const late = yield* dies(Fiber.join(straggler)) + expect(late).toBeInstanceOf(Db.AfterCommitUnavailable) + }), + ) +}) diff --git a/src/database/database.test-d.ts b/src/database/database.test-d.ts index c77a281..f684a71 100644 --- a/src/database/database.test-d.ts +++ b/src/database/database.test-d.ts @@ -1,6 +1,6 @@ -// Type-level tests: rows from `run`, `requireTransaction`, and what `transaction` removes. +// Type-level tests: rows from `run`, `requireTransaction`, `afterCommit`, and what `transaction` removes. -import { Effect, Schema } from "effect" +import { Context, Effect, Schema } from "effect" import { expectTypeOf } from "expect-type" import * as CH from "../ch/index" import * as PG from "../postgres" @@ -58,6 +58,16 @@ expectTypeOf(Db.transaction(Effect.fail(new Domain()))).toEqualTypeOf< const unwrapped: Effect.Effect = revoke("f") void unwrapped +// `afterCommit` needs a transaction, swallows the hook's errors into its report, +// and keeps the hook's own requirements: they must be met where it is queued. +class Sms extends Context.Service Effect.Effect }>()("Sms") {} +const sendLater = Db.afterCommit(Effect.flatMap(Effect.service(Sms), (sms) => sms.send("hi"))) +expectTypeOf(sendLater).toEqualTypeOf>() +expectTypeOf(Db.transaction(sendLater)).toEqualTypeOf>() +// @ts-expect-error Transaction is still required +const notQueued: Effect.Effect = sendLater +void notQueued + // Settings are typed: an unknown isolation level does not compile. // @ts-expect-error not an isolation level void Db.transaction(Effect.void, { isolationLevel: "read uncommitted" }) diff --git a/src/database/database.ts b/src/database/database.ts index aaf89d4..2ea9528 100644 --- a/src/database/database.ts +++ b/src/database/database.ts @@ -7,7 +7,7 @@ // statements outliving their transaction, and contention retry. See // `design/transactions.md`. -import { Cause, Context, Effect, Exit, Layer, Option, Schedule, Schema } from "effect" +import { Cause, Context, Effect, ErrorReporter, Exit, Layer, Option, Schedule, Schema } from "effect" import { dual } from "effect/Function" import * as SqlClient from "effect/sql/SqlClient" import { isSqlError } from "effect/sql/SqlError" @@ -20,6 +20,7 @@ import type { CHQuery, NeedsSelect } from "../ch/query" import type { CHUnionQuery, QueryParams } from "../ch/union" import type { ParamsSatisfied } from "../ch/expr" import { + AfterCommitUnavailable, DatabaseError, TransactionClosed, TransactionCommitFailed, @@ -164,11 +165,22 @@ export interface FromSqlClientOptions { readonly requireTenantScope?: boolean } +// Work queued by `afterCommit`, with the context it was queued in. +interface Hook { + readonly effect: Effect.Effect + readonly context: Context.Context +} + // Whether the transaction a fiber's context belongs to is still open. Keyed by // client so a closed transaction on one database does not flag another. +// `hooks` is this level's `afterCommit` queue: the outermost transaction runs +// it after COMMIT, a savepoint hands it to its parent when it is kept. +// `undefined` when no outer `transaction` of this client will run it (the +// transaction was opened by a plain `sql.withTransaction`). interface OpenState { open: boolean readonly client: SqlClient.SqlClient + readonly hooks: Array | undefined } class TransactionOpen extends Context.Service()("@maple-dev/effect-orm/TransactionOpen") {} @@ -380,12 +392,15 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp } const outer = yield* Effect.serviceOption(Transaction) + const parentState = yield* Effect.serviceOption(TransactionOpen) + const parentHooks = + Option.isSome(parentState) && parentState.value.client === sql && parentState.value.open ? parentState.value.hooks : undefined const info: TransactionInfo = depth === 0 || Option.isNone(outer) ? { depth, isolationLevel: settings.isolationLevel, accessMode: settings.accessMode } : { ...outer.value, depth } const setTransaction = depth === 0 ? capabilities.setTransaction(settings) : undefined - const state: OpenState = { open: true, client: sql } + const state: OpenState = { open: true, client: sql, hooks: depth === 0 || parentHooks !== undefined ? [] : undefined } let bodyExit: Exit.Exit | undefined const inner = Effect.gen(function* () { @@ -405,8 +420,10 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp ) const exit = yield* Effect.exit(sql.withTransaction(inner).pipe(Effect.ensuring(committed(state)))) - if (Exit.isSuccess(exit)) return exit.value - return yield* restoreCause(exit.cause, bodyExit, depth) + if (Exit.isFailure(exit)) return yield* restoreCause(exit.cause, bodyExit, depth) + // Kept: a savepoint's queue joins its parent's, which commits or drops it. + if (depth > 0 && parentHooks !== undefined && state.hooks !== undefined) parentHooks.push(...state.hooks) + return [exit.value, depth === 0 ? state.hooks : undefined] as const }).pipe( Effect.withSpan("effect_orm.transaction", { // Semconv defines no transaction attributes, and `db.*` is reserved for it. @@ -415,6 +432,10 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp ...(options.accessMode !== undefined && { "effect_orm.transaction.access_mode": options.accessMode }), }, }), + // After COMMIT, outside the transaction's span: the hooks are not part of it. + Effect.flatMap(([value, hooks]) => + hooks === undefined || hooks.length === 0 ? Effect.succeed(value) : Effect.as(runHooks(sql, hooks), value), + ), ) as Effect.Effect> const transaction: DatabaseApi["transaction"] = (body, options = {}) => @@ -425,6 +446,51 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp return { dialect, run, query, execute, transaction, retryContention } } +// The services that say "inside this transaction". A hook runs with the context +// it was queued in, minus these: it runs after COMMIT, on no connection of the +// transaction, and a `run` in it must not trip the closed-transaction guard. +const outsideTransaction = ( + sql: SqlClient.SqlClient, + context: Context.Context, + outside: Context.Context, +): Context.Context => { + let result = context + const keys: ReadonlyArray> = [sql.transactionService, TransactionOpen, Transaction] + for (const key of keys) { + const value = Context.getOption(outside, key) + result = Option.isSome(value) ? Context.add(result, key, value.value) : Context.omit(key)(result) + } + return result +} + +/** + * Run the hooks `afterCommit` queued, in order, after COMMIT succeeded. The + * data is already committed, so a failing hook cannot undo it and must not + * look like the transaction failed: its cause goes to the `ErrorReporter`s in + * context and to an error log, and the next hook still runs. + */ +const runHooks = (sql: SqlClient.SqlClient, hooks: ReadonlyArray): Effect.Effect => + Effect.flatMap(Effect.context(), (outside) => + Effect.forEach( + hooks, + (hook, index) => + Effect.exit(Effect.provideContext(hook.effect, outsideTransaction(sql, hook.context, outside))).pipe( + Effect.flatMap((exit) => + Exit.isSuccess(exit) + ? Effect.void + : ErrorReporter.report(exit.cause).pipe( + Effect.andThen( + Effect.logError("afterCommit hook failed after the transaction committed", exit.cause).pipe( + Effect.annotateLogs({ "effect_orm.after_commit.index": index }), + ), + ), + ), + ), + ), + { discard: true }, + ), + ).pipe(Effect.withSpan("effect_orm.after_commit", { attributes: { "effect_orm.after_commit.count": hooks.length } })) + /** * Turn what came out of `withTransaction` back into what the caller should see. * Effect dies on a failed COMMIT or ROLLBACK, and a dying ROLLBACK replaces the @@ -519,6 +585,51 @@ export const execute = (statement: StatementInput): Effect.Effect(self: Effect.Effect): Effect.Effect => Effect.andThen(Effect.service(Transaction), self) +/** + * Queue `effect` to run after the outermost transaction commits, and never if + * it rolls back. For work that must not run inside a transaction: an SMS, an + * HTTP call, a job enqueue. It adds `Transaction` to the requirements, like + * `requireTransaction`, so a helper deep in the call stack can queue work + * without knowing where the transaction starts. + * + * - Hooks run in the order they were queued, one at a time, after COMMIT and + * outside the transaction, with the context they were queued in. + * - A savepoint that rolls back drops what it queued; one that is kept hands + * it to the outer transaction. A `retry` attempt that fails drops its hooks. + * - A hook that fails does not fail the transaction, which has committed: its + * cause goes to the `ErrorReporter`s in context and to an error log, and the + * next hook still runs. Handle errors in the hook when the caller must know. + * - At most once: a crash or interruption after COMMIT skips the hooks not yet + * run. When the work must happen, write it to an outbox table in the + * transaction instead. + * + * Inside a transaction that no `transaction` call will commit (one opened by a + * plain `sql.withTransaction`), or after the transaction ended, it dies with + * `AfterCommitUnavailable` rather than drop the work. + * + * ```ts + * yield* Db.afterCommit(sms.send(phone, text)) + * ``` + */ +export const afterCommit = (effect: Effect.Effect): Effect.Effect => + Effect.gen(function* () { + yield* Transaction + const state = yield* Effect.serviceOption(TransactionOpen) + const unavailable = (reason: string) => + Effect.die(new AfterCommitUnavailable({ message: `afterCommit cannot queue work: ${reason}` })) + if (Option.isNone(state)) return yield* unavailable("no transaction opened by Db.transaction is open") + if (!state.value.open) { + return yield* unavailable("its transaction has ended; join fibers forked inside a transaction before it returns") + } + if (state.value.hooks === undefined) { + return yield* unavailable("the outermost transaction was opened by sql.withTransaction, which never runs queued work") + } + state.value.hooks.push({ + effect: effect as Effect.Effect, + context: (yield* Effect.context()) as Context.Context, + }) + }) + /** * `transaction` on the `Database` in context. Data-first or pipeable, so it * also fits as an `Effect.fn` pipe argument: diff --git a/src/database/errors.ts b/src/database/errors.ts index 786ab5a..fc0ee2d 100644 --- a/src/database/errors.ts +++ b/src/database/errors.ts @@ -65,6 +65,18 @@ export class TransactionClosed extends Schema.TaggedError()(" sql: Schema.String, }) {} +/** + * `afterCommit` had nowhere to queue its work: no transaction opened by + * `transaction` is open in this fiber, the transaction already ended, or the + * outermost transaction was opened by a plain `sql.withTransaction`, which + * would never run the work. It is a defect, so the work is never dropped + * silently. + */ +export class AfterCommitUnavailable extends Schema.TaggedError()( + "@maple-dev/effect-orm/AfterCommitUnavailable", + { message: Schema.String }, +) {} + /** * A database with `requireTenantScope` was handed a compiled statement that * reads or writes across tenants without saying so: a forgotten tenant From 965b699bc8f8083cb1fc0003b8f414470a8856e7 Mon Sep 17 00:00:00 2001 From: Taras Kornichuk Date: Sun, 25 Oct 2026 17:05:21 +0200 Subject: [PATCH 2/4] docs(database): document afterCommit and its failure policy --- design/transactions.md | 26 ++++++++++++++++++++++++ docs/database.md | 45 +++++++++++++++++++++++++++++++++++++++++- docs/reference.md | 4 ++-- 3 files changed, 72 insertions(+), 3 deletions(-) diff --git a/design/transactions.md b/design/transactions.md index b8ffda5..7ad863a 100644 --- a/design/transactions.md +++ b/design/transactions.md @@ -646,3 +646,29 @@ 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`. + +## 10. `afterCommit` (issue #73) + +`Db.afterCommit(effect)` queues work that must wait for COMMIT: a vendor call, a job enqueue. +Each `transaction` level keeps a queue on its private open-state service. The outermost level +runs its queue after COMMIT; a savepoint that is kept appends its queue to its parent's, and +one that rolls back drops it. A hook captures the context it was queued in, so its own +requirements are met where it was queued; the transaction services +(`sql.transactionService`, the open-state, `Transaction`) are replaced by the ones outside the +transaction, so a statement in a hook runs on its own connection. + +Decided: + +- **A failing hook does not fail the transaction.** The data is committed; a typed failure at + that point would invite a retry that writes twice. The cause goes to + `ErrorReporter.report` (Effect's channel for observed failures) and `Effect.logError`, and + the next hook still runs. A caller that must know handles the error inside the hook. +- **Nowhere to queue is a defect** (`AfterCommitUnavailable`), like `TransactionClosed`: a + hand-provided `Transaction`, an ended transaction, or an outermost transaction opened by a + plain `sql.withTransaction` would drop the work without a sign. +- **At most once.** Hooks are not run uninterruptibly and are not persisted. Guaranteed + delivery is an outbox table written inside the transaction, which needs no library support. + +Not done: dialects other than Postgres (SQLite, MySQL, D1 `batch()`) have no transactions on +main yet. Their `transaction` will run the same queue without change, because the queue lives +in `once`, not in a dialect. diff --git a/docs/database.md b/docs/database.md index 2e0cccd..009596b 100644 --- a/docs/database.md +++ b/docs/database.md @@ -89,6 +89,7 @@ Calling `withdraw(1, 30)` outside `transfer` does not compile: `requireTransacti | `execute(statement)` | Run a statement and discard its rows | | `transaction(options?)` | Run an effect in a transaction; data-first or pipeable | | `requireTransaction` | Mark an effect as correct only inside a transaction | +| `afterCommit(effect)` | Queue an effect to run after the outermost transaction commits, never after a rollback | | `retryContention(options?)` | Re-run on serialization failure or deadlock | | `Transaction` | The open transaction (`depth`, `isolationLevel`, `accessMode`) | | `isContention(error)` | Whether an error is a serialization failure or deadlock | @@ -196,6 +197,47 @@ Effect's own `Effect.tx` treats `Effect.Transaction`. A helper that works either most reads, needs no marker. Inside a transaction, `yield* Db.Transaction` gives its `depth` and settings. +### After commit + +A vendor call (an SMS, an HTTP request, a job enqueue) must not run inside a transaction: it +holds the connection while it waits, and it cannot be rolled back. `afterCommit` queues the +call and runs it only after COMMIT succeeds: + +```ts +const bookTour = Effect.fn("bookTour")(function* (tour: NewTour) { + yield* Db.execute(Db.sql`INSERT INTO tours VALUES (${tour.id}, ${tour.phone})`) + yield* Db.afterCommit(sms.send(tour.phone, "Your tour is booked")) +}, Db.requireTransaction) + +Db.transaction(bookTour(tour)) // the SMS goes out after COMMIT, or not at all +``` + +Like `requireTransaction`, `afterCommit` adds `Transaction` to the requirements, so a helper +deep in the call stack can queue work without knowing where the transaction starts. The +requirements of the queued effect stay in `R`: they must be met where it is queued. + +- **Order.** Hooks run one at a time, in the order they were queued, after COMMIT and outside + the transaction. A statement in a hook runs on its own connection, not the transaction's. +- **Rollback.** A rolled-back transaction runs nothing. A savepoint that rolls back drops what + it queued; a savepoint that is kept hands its hooks to the outer transaction. With `retry`, + only the attempt that commits runs its hooks. +- **Context.** A hook runs with the services of the place it was queued, minus the + transaction. +- **Failures.** A hook that fails or dies does not fail the transaction, because the data is + already committed and a caller that retries would write it twice. The hook's cause goes to + the `ErrorReporter`s in context (`Effect.withErrorReporting` uses the same ones) and to an + error log, and the next hook still runs. When the caller must know that the call failed, + handle the error inside the hook. +- **At most once.** A crash or an interruption after COMMIT skips the hooks not yet run. When + the work must happen, write it to an outbox table inside the transaction instead. +- **Tracing.** Hooks run in an `effect_orm.after_commit` span, outside the transaction's span. + +`afterCommit` dies with `AfterCommitUnavailable` when no `transaction` would run the work: a +`Transaction` provided by hand, a transaction that has already ended (a fiber forked inside it +and not joined), or an outermost transaction opened by a plain `sql.withTransaction`. A savepoint +made by a plain `sql.withTransaction` inside a `Db.transaction` does not drop hooks when it rolls +back; open nested transactions with `Db.transaction`. + ### Settings | Option | Postgres | @@ -227,7 +269,8 @@ from `body` stay defects. `TransactionClosed` is a defect, not an error to handle. A fiber forked inside a transaction inherits its connection. If it runs a statement after the transaction ended, that statement would run on a connection already back in the pool, so `run`, `query` and `execute` die with -`TransactionClosed` instead. Join what you fork before the body returns. +`TransactionClosed` instead. Join what you fork before the body returns. `AfterCommitUnavailable` is a defect for the same +reason: work queued where nothing will run it would be lost without a sign. ### Contention retry diff --git a/docs/reference.md b/docs/reference.md index abc2d46..f4e5304 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -511,8 +511,8 @@ placeholders, double-quoted identifiers. | `ClickHouseStatementFromString`, `splitTerminalClauses`, `maskLiteralsAndComments` | `/sql` | | Schema tooling: `isSchemaObject`, `makeSnapshot`, `renderSchema`, `diffSchemas`, `diffPgSchemas`, `fromDrizzleSnapshot`, snapshot and entity schemas (`PgEnumEntity`, `PgUniqueEntity`, …), `Hint` | `/schema` | | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors | `/migrate` | -| `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `retryContention`, `Transaction` | `/database` | -| `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | +| `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` | +| `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed`, `AfterCommitUnavailable` and the other transaction errors | `/database` | | `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` | | `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` | | `defineConfig`, `generate`, `check`, `pull`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit` | From ec94fd2a58c16fe265a7584ebf86ef98bad93527 Mon Sep 17 00:00:00 2001 From: Taras Kornichuk Date: Sun, 25 Oct 2026 17:05:21 +0200 Subject: [PATCH 3/4] chore: changeset for afterCommit --- .changeset/after-commit.md | 5 +++++ 1 file changed, 5 insertions(+) create mode 100644 .changeset/after-commit.md diff --git a/.changeset/after-commit.md b/.changeset/after-commit.md new file mode 100644 index 0000000..eefa802 --- /dev/null +++ b/.changeset/after-commit.md @@ -0,0 +1,5 @@ +--- +"@maple-dev/effect-orm": minor +--- + +`Db.afterCommit(effect)` queues work, such as an SMS or an HTTP call, to run after the outermost transaction commits, and never after a rollback. Like `requireTransaction`, it adds `Transaction` to the requirements. Hooks run in order, outside the transaction, with the services of the place they were queued. A savepoint that rolls back drops its hooks, and a failed `retry` attempt drops its hooks. A failing hook does not fail the committed transaction: its cause goes to the `ErrorReporter`s in context and to an error log, and the next hook still runs. Where nothing would run the work (a `Transaction` provided by hand, a transaction that has ended, or one opened by a plain `sql.withTransaction`), `afterCommit` dies with the new `AfterCommitUnavailable`. From 72e620465a82b362c8ebd211e3efd89b5755584f Mon Sep 17 00:00:00 2001 From: Makisuo Date: Mon, 26 Oct 2026 00:32:01 +0100 Subject: [PATCH 4/4] refactor(database): build afterCommit on Effect's SqlClient.afterCommit Effect 4.0.6 ships SqlClient.afterCommit with the same queueing rules: after the outermost COMMIT, dropped on rollback, a kept savepoint hands its callbacks to its parent, and a defect after the transaction ended. Db.afterCommit now delegates to it and keeps only what Effect lacks: Transaction in R, and a failing hook reported to ErrorReporter and the log instead of failing the committed transaction. The hook queue on OpenState, runHooks and AfterCommitUnavailable go. A plain sql.withTransaction as the outer transaction now runs the hooks instead of dying, since Effect queues them there too. Co-Authored-By: Claude Opus 5.5 --- .changeset/after-commit.md | 2 +- design/transactions.md | 31 +++----- docs/database.md | 19 +++-- docs/reference.md | 2 +- src/database.ts | 1 - src/database/after-commit.test.ts | 20 ++--- src/database/database.ts | 124 ++++++------------------------ src/database/errors.ts | 12 --- 8 files changed, 59 insertions(+), 152 deletions(-) diff --git a/.changeset/after-commit.md b/.changeset/after-commit.md index eefa802..75c8320 100644 --- a/.changeset/after-commit.md +++ b/.changeset/after-commit.md @@ -2,4 +2,4 @@ "@maple-dev/effect-orm": minor --- -`Db.afterCommit(effect)` queues work, such as an SMS or an HTTP call, to run after the outermost transaction commits, and never after a rollback. Like `requireTransaction`, it adds `Transaction` to the requirements. Hooks run in order, outside the transaction, with the services of the place they were queued. A savepoint that rolls back drops its hooks, and a failed `retry` attempt drops its hooks. A failing hook does not fail the committed transaction: its cause goes to the `ErrorReporter`s in context and to an error log, and the next hook still runs. Where nothing would run the work (a `Transaction` provided by hand, a transaction that has ended, or one opened by a plain `sql.withTransaction`), `afterCommit` dies with the new `AfterCommitUnavailable`. +`Db.afterCommit(effect)` queues work, such as an SMS or an HTTP call, to run after the outermost transaction commits, and never after a rollback. It delegates to Effect's `SqlClient.afterCommit` and, like `requireTransaction`, adds `Transaction` to the requirements, so queueing outside a transaction does not compile. A failing hook does not fail the committed transaction: its cause goes to the `ErrorReporter`s in context and to an error log, and the next hook still runs. diff --git a/design/transactions.md b/design/transactions.md index e90d7c9..4fa09d8 100644 --- a/design/transactions.md +++ b/design/transactions.md @@ -654,25 +654,18 @@ What landed differently from the sections above: ## 10. `afterCommit` (issue #73) `Db.afterCommit(effect)` queues work that must wait for COMMIT: a vendor call, a job enqueue. -Each `transaction` level keeps a queue on its private open-state service. The outermost level -runs its queue after COMMIT; a savepoint that is kept appends its queue to its parent's, and -one that rolls back drops it. A hook captures the context it was queued in, so its own -requirements are met where it was queued; the transaction services -(`sql.transactionService`, the open-state, `Transaction`) are replaced by the ones outside the -transaction, so a statement in a hook runs on its own connection. - -Decided: +Effect 4.0.6 ships `SqlClient.afterCommit`, which already runs callbacks after the outermost +COMMIT, drops those of a rolled-back transaction or savepoint, hands a kept savepoint's callbacks +to its parent, and dies when queued after the transaction ended. `Db.afterCommit` delegates to +it on the client of the open `Db.transaction` and adds only: +- **`Transaction` in `R`.** Effect's version runs the work at once outside a transaction; here + that does not compile, like `requireTransaction`. - **A failing hook does not fail the transaction.** The data is committed; a typed failure at that point would invite a retry that writes twice. The cause goes to - `ErrorReporter.report` (Effect's channel for observed failures) and `Effect.logError`, and - the next hook still runs. A caller that must know handles the error inside the hook. -- **Nowhere to queue is a defect** (`AfterCommitUnavailable`), like `TransactionClosed`: a - hand-provided `Transaction`, an ended transaction, or an outermost transaction opened by a - plain `sql.withTransaction` would drop the work without a sign. -- **At most once.** Hooks are not run uninterruptibly and are not persisted. Guaranteed - delivery is an outbox table written inside the transaction, which needs no library support. - -Not done: dialects other than Postgres (SQLite, MySQL, D1 `batch()`) have no transactions on -main yet. Their `transaction` will run the same queue without change, because the queue lives -in `once`, not in a dialect. + `ErrorReporter.report` and `Effect.logError`, and the next hook still runs. +- **The hook's context drops the open-state and `Transaction`** as Effect drops its own + transaction service, so a statement in a hook does not trip the closed-transaction guard. + +At most once: hooks are not persisted. Guaranteed delivery is an outbox table written inside +the transaction, which needs no library support. diff --git a/docs/database.md b/docs/database.md index 2d22f70..bf14d00 100644 --- a/docs/database.md +++ b/docs/database.md @@ -216,6 +216,9 @@ Like `requireTransaction`, `afterCommit` adds `Transaction` to the requirements, deep in the call stack can queue work without knowing where the transaction starts. The requirements of the queued effect stay in `R`: they must be met where it is queued. +Effect's own `SqlClient.afterCommit` does the queueing; `afterCommit` adds the `Transaction` +requirement and the failure policy below. + - **Order.** Hooks run one at a time, in the order they were queued, after COMMIT and outside the transaction. A statement in a hook runs on its own connection, not the transaction's. - **Rollback.** A rolled-back transaction runs nothing. A savepoint that rolls back drops what @@ -228,15 +231,12 @@ requirements of the queued effect stay in `R`: they must be met where it is queu the `ErrorReporter`s in context (`Effect.withErrorReporting` uses the same ones) and to an error log, and the next hook still runs. When the caller must know that the call failed, handle the error inside the hook. -- **At most once.** A crash or an interruption after COMMIT skips the hooks not yet run. When - the work must happen, write it to an outbox table inside the transaction instead. -- **Tracing.** Hooks run in an `effect_orm.after_commit` span, outside the transaction's span. +- **At most once.** A crash before the hooks run skips them. Hooks run before `transaction` + returns and uninterruptibly, so keep them short and fork long-running work. When the work + must happen, write it to an outbox table inside the transaction instead. -`afterCommit` dies with `AfterCommitUnavailable` when no `transaction` would run the work: a -`Transaction` provided by hand, a transaction that has already ended (a fiber forked inside it -and not joined), or an outermost transaction opened by a plain `sql.withTransaction`. A savepoint -made by a plain `sql.withTransaction` inside a `Db.transaction` does not drop hooks when it rolls -back; open nested transactions with `Db.transaction`. +`afterCommit` dies when nothing would run the work: a `Transaction` provided by hand, or a +transaction that has already ended (a fiber forked inside it and not joined). ### Settings @@ -269,8 +269,7 @@ from `body` stay defects. `TransactionClosed` is a defect, not an error to handle. A fiber forked inside a transaction inherits its connection. If it runs a statement after the transaction ended, that statement would run on a connection already back in the pool, so `run`, `query` and `execute` die with -`TransactionClosed` instead. Join what you fork before the body returns. `AfterCommitUnavailable` is a defect for the same -reason: work queued where nothing will run it would be lost without a sign. +`TransactionClosed` instead. Join what you fork before the body returns. ### Contention retry diff --git a/docs/reference.md b/docs/reference.md index 4f64efa..fe8f4eb 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -513,7 +513,7 @@ placeholders, double-quoted identifiers. | Schema tooling: `isSchemaObject`, `makeSnapshot`, `renderSchema`, `diffSchemas`, `diffPgSchemas`, `fromDrizzleSnapshot`, snapshot and entity schemas (`PgEnumEntity`, `PgUniqueEntity`, …), `Hint`; for SQLite, `sqliteEntitiesOf`, `renderSqliteSchema`, `renderSqliteCreateTable`, `renderSqliteCreateIndex`, `diffSqliteSchemas`, `fromDrizzleSqliteSnapshot`, `SqliteMigrationOp`, `renderSqliteOp`, `renderSqliteMigrationSql`, `labelOfSqlite`, `rebuildName`, the `SqliteSnapshot` schema and the `SqliteTableEntity`, `SqliteColumnEntity`, `SqliteIndexEntity`, `SqliteUniqueEntity` schemas | `/schema` | | `run`, `status`, `verify`, `baseline`, `MigrationDriver` and the migrate errors | `/migrate` | | `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/database` | -| `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed`, `AfterCommitUnavailable` and the other transaction errors | `/database` | +| `DatabaseError`, `TransactionCommitFailed`, `TransactionRollbackFailed` and the other transaction errors | `/database` | | `withQuerySummary`, `QuerySummary` — the `db.query.summary` of the statements an effect runs, for spans | `/database` | | `ClickhouseHttpClient`, `layer`, `make`, `ClickhouseHttpClientConfig` — a ClickHouse `SqlClient` over `fetch`, for Workers | `/clickhouse-http` | | `defineConfig`, `generate`, `check`, `pull`, `loadSchema`, `readMigrations`, `analyze`, `KitError` | `/kit` | diff --git a/src/database.ts b/src/database.ts index 865972e..dc56aab 100644 --- a/src/database.ts +++ b/src/database.ts @@ -30,7 +30,6 @@ export { type TransactionOptions, } from "./database/database" export { - AfterCommitUnavailable, DatabaseError, TenantScopeViolation, TransactionClosed, diff --git a/src/database/after-commit.test.ts b/src/database/after-commit.test.ts index 4562f0d..0bf789e 100644 --- a/src/database/after-commit.test.ts +++ b/src/database/after-commit.test.ts @@ -164,25 +164,27 @@ layer(Live, { excludeTestServices: true })("afterCommit on PGlite", (it) => { }), ) - it.effect("dies with AfterCommitUnavailable rather than drop work no transaction will run", () => + it.effect("runs under a plain sql.withTransaction too; dies where nothing will run it", () => Effect.gen(function* () { const sql = yield* SqlClient.SqlClient - const work = Effect.void + const ran = yield* Ref.make(0) + const work = Ref.update(ran, (n) => n + 1) - // The outermost transaction is a plain sql.withTransaction: nothing would run the hook. - const foreign = yield* dies(sql.withTransaction(Db.transaction(Db.afterCommit(work)))) - expect(foreign).toBeInstanceOf(Db.AfterCommitUnavailable) + // Effect's own transaction queues it: it runs once the outer COMMIT succeeds. + yield* sql.withTransaction( + Db.transaction(Db.afterCommit(work)).pipe(Effect.andThen(Effect.flatMap(Ref.get(ran), (n) => Effect.sync(() => expect(n).toBe(0))))), + ) + expect(yield* Ref.get(ran)).toBe(1) // Transaction provided by hand, with no transaction behind it. - const fake = yield* dies(Db.afterCommit(work).pipe(Effect.provideService(Db.Transaction, { depth: 0, isolationLevel: undefined, accessMode: undefined }))) - expect(fake).toBeInstanceOf(Db.AfterCommitUnavailable) + yield* dies(Db.afterCommit(work).pipe(Effect.provideService(Db.Transaction, { depth: 0, isolationLevel: undefined, accessMode: undefined }))) // A fiber forked in the transaction that queues work after it ended. const release = yield* Deferred.make() const straggler = yield* Db.transaction(Deferred.await(release).pipe(Effect.andThen(Db.afterCommit(work)), Effect.forkDetach)) yield* Deferred.succeed(release, undefined) - const late = yield* dies(Fiber.join(straggler)) - expect(late).toBeInstanceOf(Db.AfterCommitUnavailable) + yield* dies(Fiber.join(straggler)) + expect(yield* Ref.get(ran)).toBe(1) }), ) }) diff --git a/src/database/database.ts b/src/database/database.ts index 746e40a..3a94eaa 100644 --- a/src/database/database.ts +++ b/src/database/database.ts @@ -20,7 +20,6 @@ import type { CHQuery, NeedsSelect } from "../ch/query" import type { CHUnionQuery, QueryParams } from "../ch/union" import type { ParamsSatisfied } from "../ch/expr" import { - AfterCommitUnavailable, DatabaseError, TransactionClosed, TransactionCommitFailed, @@ -165,22 +164,11 @@ export interface FromSqlClientOptions { readonly requireTenantScope?: boolean } -// Work queued by `afterCommit`, with the context it was queued in. -interface Hook { - readonly effect: Effect.Effect - readonly context: Context.Context -} - // Whether the transaction a fiber's context belongs to is still open. Keyed by // client so a closed transaction on one database does not flag another. -// `hooks` is this level's `afterCommit` queue: the outermost transaction runs -// it after COMMIT, a savepoint hands it to its parent when it is kept. -// `undefined` when no outer `transaction` of this client will run it (the -// transaction was opened by a plain `sql.withTransaction`). interface OpenState { open: boolean readonly client: SqlClient.SqlClient - readonly hooks: Array | undefined } class TransactionOpen extends Context.Service()("@maple-dev/effect-orm/TransactionOpen") {} @@ -392,15 +380,12 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp } const outer = yield* Effect.serviceOption(Transaction) - const parentState = yield* Effect.serviceOption(TransactionOpen) - const parentHooks = - Option.isSome(parentState) && parentState.value.client === sql && parentState.value.open ? parentState.value.hooks : undefined const info: TransactionInfo = depth === 0 || Option.isNone(outer) ? { depth, isolationLevel: settings.isolationLevel, accessMode: settings.accessMode } : { ...outer.value, depth } const setTransaction = depth === 0 ? capabilities.setTransaction(settings) : undefined - const state: OpenState = { open: true, client: sql, hooks: depth === 0 || parentHooks !== undefined ? [] : undefined } + const state: OpenState = { open: true, client: sql } let bodyExit: Exit.Exit | undefined const inner = Effect.gen(function* () { @@ -420,10 +405,8 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp ) const exit = yield* Effect.exit(sql.withTransaction(inner).pipe(Effect.ensuring(committed(state)))) - if (Exit.isFailure(exit)) return yield* restoreCause(exit.cause, bodyExit, depth) - // Kept: a savepoint's queue joins its parent's, which commits or drops it. - if (depth > 0 && parentHooks !== undefined && state.hooks !== undefined) parentHooks.push(...state.hooks) - return [exit.value, depth === 0 ? state.hooks : undefined] as const + if (Exit.isSuccess(exit)) return exit.value + return yield* restoreCause(exit.cause, bodyExit, depth) }).pipe( Effect.withSpan("effect_orm.transaction", { // Semconv defines no transaction attributes, and `db.*` is reserved for it. @@ -432,10 +415,6 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp ...(options.accessMode !== undefined && { "effect_orm.transaction.access_mode": options.accessMode }), }, }), - // After COMMIT, outside the transaction's span: the hooks are not part of it. - Effect.flatMap(([value, hooks]) => - hooks === undefined || hooks.length === 0 ? Effect.succeed(value) : Effect.as(runHooks(sql, hooks), value), - ), ) as Effect.Effect> const transaction: DatabaseApi["transaction"] = (body, options = {}) => @@ -446,51 +425,6 @@ export const fromSqlClient = (sql: SqlClient.SqlClient, options: FromSqlClientOp return { dialect, run, query, execute, transaction, retryContention } } -// The services that say "inside this transaction". A hook runs with the context -// it was queued in, minus these: it runs after COMMIT, on no connection of the -// transaction, and a `run` in it must not trip the closed-transaction guard. -const outsideTransaction = ( - sql: SqlClient.SqlClient, - context: Context.Context, - outside: Context.Context, -): Context.Context => { - let result = context - const keys: ReadonlyArray> = [sql.transactionService, TransactionOpen, Transaction] - for (const key of keys) { - const value = Context.getOption(outside, key) - result = Option.isSome(value) ? Context.add(result, key, value.value) : Context.omit(key)(result) - } - return result -} - -/** - * Run the hooks `afterCommit` queued, in order, after COMMIT succeeded. The - * data is already committed, so a failing hook cannot undo it and must not - * look like the transaction failed: its cause goes to the `ErrorReporter`s in - * context and to an error log, and the next hook still runs. - */ -const runHooks = (sql: SqlClient.SqlClient, hooks: ReadonlyArray): Effect.Effect => - Effect.flatMap(Effect.context(), (outside) => - Effect.forEach( - hooks, - (hook, index) => - Effect.exit(Effect.provideContext(hook.effect, outsideTransaction(sql, hook.context, outside))).pipe( - Effect.flatMap((exit) => - Exit.isSuccess(exit) - ? Effect.void - : ErrorReporter.report(exit.cause).pipe( - Effect.andThen( - Effect.logError("afterCommit hook failed after the transaction committed", exit.cause).pipe( - Effect.annotateLogs({ "effect_orm.after_commit.index": index }), - ), - ), - ), - ), - ), - { discard: true }, - ), - ).pipe(Effect.withSpan("effect_orm.after_commit", { attributes: { "effect_orm.after_commit.count": hooks.length } })) - /** * Turn what came out of `withTransaction` back into what the caller should see. * Effect fails with the `SqlError` of a failed COMMIT (or dies, when cleanup @@ -596,25 +530,14 @@ export const requireTransaction = (self: Effect.Effect): Effec /** * Queue `effect` to run after the outermost transaction commits, and never if - * it rolls back. For work that must not run inside a transaction: an SMS, an - * HTTP call, a job enqueue. It adds `Transaction` to the requirements, like - * `requireTransaction`, so a helper deep in the call stack can queue work - * without knowing where the transaction starts. + * it rolls back: an SMS, an HTTP call, a job enqueue. Effect's own + * `SqlClient.afterCommit` does the queueing (savepoints, ordering, rollback); + * this adds `Transaction` to the requirements, like `requireTransaction`, so + * work queued outside a transaction does not compile instead of running at once. * - * - Hooks run in the order they were queued, one at a time, after COMMIT and - * outside the transaction, with the context they were queued in. - * - A savepoint that rolls back drops what it queued; one that is kept hands - * it to the outer transaction. A `retry` attempt that fails drops its hooks. - * - A hook that fails does not fail the transaction, which has committed: its - * cause goes to the `ErrorReporter`s in context and to an error log, and the - * next hook still runs. Handle errors in the hook when the caller must know. - * - At most once: a crash or interruption after COMMIT skips the hooks not yet - * run. When the work must happen, write it to an outbox table in the - * transaction instead. - * - * Inside a transaction that no `transaction` call will commit (one opened by a - * plain `sql.withTransaction`), or after the transaction ended, it dies with - * `AfterCommitUnavailable` rather than drop the work. + * A hook that fails does not fail the committed transaction: its cause goes to + * the `ErrorReporter`s in context and to an error log. Hooks run at most once; + * work that must happen belongs in an outbox table written in the transaction. * * ```ts * yield* Db.afterCommit(sms.send(phone, text)) @@ -624,19 +547,22 @@ export const afterCommit = (effect: Effect.Effect): Effect.Eff Effect.gen(function* () { yield* Transaction const state = yield* Effect.serviceOption(TransactionOpen) - const unavailable = (reason: string) => - Effect.die(new AfterCommitUnavailable({ message: `afterCommit cannot queue work: ${reason}` })) - if (Option.isNone(state)) return yield* unavailable("no transaction opened by Db.transaction is open") - if (!state.value.open) { - return yield* unavailable("its transaction has ended; join fibers forked inside a transaction before it returns") - } - if (state.value.hooks === undefined) { - return yield* unavailable("the outermost transaction was opened by sql.withTransaction, which never runs queued work") + if (Option.isNone(state)) { + return yield* Effect.die(new Error("Db.afterCommit: no transaction opened by Db.transaction is open")) } - state.value.hooks.push({ - effect: effect as Effect.Effect, - context: (yield* Effect.context()) as Context.Context, - }) + yield* state.value.client.afterCommit( + effect.pipe( + Effect.catchCause((cause) => + ErrorReporter.report(cause).pipe( + Effect.andThen(Effect.logError("afterCommit hook failed after the transaction committed", cause)), + ), + ), + Effect.asVoid, + // Effect drops its own transaction from the hook's context; drop ours too, + // so a statement in the hook does not trip the closed-transaction guard. + Effect.updateContext((context: Context.Context) => Context.omit(TransactionOpen, Transaction)(context) as Context.Context), + ), + ) }) /** diff --git a/src/database/errors.ts b/src/database/errors.ts index fc0ee2d..786ab5a 100644 --- a/src/database/errors.ts +++ b/src/database/errors.ts @@ -65,18 +65,6 @@ export class TransactionClosed extends Schema.TaggedError()(" sql: Schema.String, }) {} -/** - * `afterCommit` had nowhere to queue its work: no transaction opened by - * `transaction` is open in this fiber, the transaction already ended, or the - * outermost transaction was opened by a plain `sql.withTransaction`, which - * would never run the work. It is a defect, so the work is never dropped - * silently. - */ -export class AfterCommitUnavailable extends Schema.TaggedError()( - "@maple-dev/effect-orm/AfterCommitUnavailable", - { message: Schema.String }, -) {} - /** * A database with `requireTenantScope` was handed a compiled statement that * reads or writes across tenants without saying so: a forgotten tenant