diff --git a/.changeset/after-commit.md b/.changeset/after-commit.md new file mode 100644 index 0000000..75c8320 --- /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. 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 854ce61..4fa09d8 100644 --- a/design/transactions.md +++ b/design/transactions.md @@ -650,3 +650,22 @@ 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. +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` 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 9208a6b..bf14d00 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. + +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 + 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 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 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 | Option | Postgres | diff --git a/docs/reference.md b/docs/reference.md index ee49eb1..fe8f4eb 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -512,7 +512,7 @@ placeholders, double-quoted identifiers. | `ClickHouseStatementFromString`, `splitTerminalClauses`, `maskLiteralsAndComments` | `/sql` | | 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`, `retryContention`, `Transaction` | `/database` | +| `Database`, `run`, `sql`, `query`, `execute`, `transaction`, `requireTransaction`, `afterCommit`, `retryContention`, `Transaction` | `/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` | diff --git a/src/database.ts b/src/database.ts index ba66164..dc56aab 100644 --- a/src/database.ts +++ b/src/database.ts @@ -8,6 +8,7 @@ export { Database, Transaction, + afterCommit, execute, fromSqlClient, isContention, diff --git a/src/database/after-commit.test.ts b/src/database/after-commit.test.ts new file mode 100644 index 0000000..0bf789e --- /dev/null +++ b/src/database/after-commit.test.ts @@ -0,0 +1,190 @@ +// `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("runs under a plain sql.withTransaction too; dies where nothing will run it", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient + const ran = yield* Ref.make(0) + const work = Ref.update(ran, (n) => n + 1) + + // 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. + 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) + yield* dies(Fiber.join(straggler)) + expect(yield* Ref.get(ran)).toBe(1) + }), + ) +}) 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 116e582..3a94eaa 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" @@ -528,6 +528,43 @@ 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: 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. + * + * 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)) + * ``` + */ +export const afterCommit = (effect: Effect.Effect): Effect.Effect => + Effect.gen(function* () { + yield* Transaction + const state = yield* Effect.serviceOption(TransactionOpen) + if (Option.isNone(state)) { + return yield* Effect.die(new Error("Db.afterCommit: no transaction opened by Db.transaction is open")) + } + 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), + ), + ) + }) + /** * `transaction` on the `Database` in context. Data-first or pipeable, so it * also fits as an `Effect.fn` pipe argument: