From 8906fd1a037d203fcefe3bc845d2df66a25c4388 Mon Sep 17 00:00:00 2001 From: Paul Dlug Date: Fri, 23 Oct 2026 14:00:40 -0700 Subject: [PATCH 1/2] fix(postgres): keep postgres-js transactions committable after a recovered savepoint refusal postgres-js records the first failed statement of each begin or savepoint scope and rethrows it when that scope ends, unless the statement ran through one of the driver's own nested savepoint scopes. TypeGraph recovered deliberate statement failures with raw SAVEPOINT and ROLLBACK TO statements on the outer transaction handle, so on postgres-js the caller's transaction rejected at commit with the already-recovered driver error. One savepoint primitive, runSessionSavepoint in src/backend/session-savepoint.ts, now owns savepoint control for the session atomic program, recorded-capture savepoints, and the schema-adoption probe. On a postgres-js transaction it enters the driver's own savepoint scope (Drizzle's nested transaction) inside TypeGraph's savepoint, and the transaction session routes every statement through the innermost open scope's handle while it is open. Releasing TypeGraph's savepoint also releases the driver's, which the driver never releases itself. --- .changeset/postgres-js-savepoint-recovery.md | 5 + .../execution/postgres-transaction-session.ts | 147 +++++++++++ .../drizzle/execution/session-atomic-batch.ts | 65 +++-- .../typegraph/src/backend/drizzle/postgres.ts | 40 ++- .../src/backend/session-savepoint.ts | 154 ++++++++++++ .../typegraph/src/store/recorded-capture.ts | 50 ++-- .../tests/backends/integration-test-suite.ts | 2 + .../backends/integration/adopted-evolution.ts | 31 +++ .../tests/backends/integration/index.ts | 1 + .../integration/savepoint-recovery.ts | 230 ++++++++++++++++++ .../recorded-capture-write-parity.test.ts | 3 +- .../typegraph/tests/session-savepoint.test.ts | 171 +++++++++++++ 12 files changed, 816 insertions(+), 83 deletions(-) create mode 100644 .changeset/postgres-js-savepoint-recovery.md create mode 100644 packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts create mode 100644 packages/typegraph/src/backend/session-savepoint.ts create mode 100644 packages/typegraph/tests/backends/integration/savepoint-recovery.ts create mode 100644 packages/typegraph/tests/session-savepoint.test.ts diff --git a/.changeset/postgres-js-savepoint-recovery.md b/.changeset/postgres-js-savepoint-recovery.md new file mode 100644 index 000000000..decbfa211 --- /dev/null +++ b/.changeset/postgres-js-savepoint-recovery.md @@ -0,0 +1,5 @@ +--- +"@nicia-ai/typegraph": patch +--- + +Keep a transaction committable on the postgres-js driver after TypeGraph recovers a refusal through a savepoint. postgres-js remembers the first failed statement of a transaction and rethrows it at commit unless the statement ran inside one of the driver's own savepoint scopes, so a `UniquenessError` caught inside `store.transaction(...)` after `bulkReplaceById` (or another fused write program) lost a unique key still failed the transaction at commit with the recovered driver error, and an `importGraph` whose per-row edge create was refused by the database (a deferred row losing its durable match identity) rolled back the whole import instead of reporting that row as an error. TypeGraph now runs the recovered work inside a postgres-js savepoint scope nested in its own savepoint, so the caller's transaction commits what it wrote outside the refused write on every PostgreSQL driver. A savepoint recovery that itself fails is reported as one `AggregateError` naming both failures. diff --git a/packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts b/packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts new file mode 100644 index 000000000..8ad29d297 --- /dev/null +++ b/packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts @@ -0,0 +1,147 @@ +import { CompilerInvariantError } from "../../../errors"; +import { type DriverSavepointScope } from "../../session-savepoint"; +import { + type AnyPgDatabase, + createPostgresExecutionAdapter, + getPinnedPostgresTransactionClient, + isPostgresJsClient, + type PostgresExecutionAdapterOptions, +} from "./postgres-execution"; +import type { + CompiledSqlQuery, + ExecutableSql, + PreparedSqlStatement, + SqlExecutionAdapter, +} from "./types"; + +type ScopeExecution = Readonly<{ + db: AnyPgDatabase; + operation: SqlExecutionAdapter; + compiled: SqlExecutionAdapter; +}>; + +/** + * The execution surfaces of one open PostgreSQL transaction session. + * + * `operation` keeps ordinary transaction work on Drizzle's session; `compiled` + * runs already-compiled statements on the pinned driver client. + * `enterDriverScope` is present when the driver attributes a failed statement + * to the scope whose handle issued it (postgres-js rethrows the first one when + * its `begin` or `savepoint` scope ends). While such a scope is open, both + * surfaces issue every statement through its handle: PostgreSQL places them + * inside the savepoint whichever handle sent them, and routing them through it + * keeps the driver's attribution in step with the server. + */ +type PostgresTransactionSession = Readonly<{ + operation: SqlExecutionAdapter; + compiled: SqlExecutionAdapter; + enterDriverScope: DriverSavepointScope | undefined; +}>; + +function bindScopeExecution( + db: AnyPgDatabase, + options: PostgresExecutionAdapterOptions | undefined, +): ScopeExecution { + return { + db, + operation: createPostgresExecutionAdapter(db, options), + compiled: createPostgresExecutionAdapter(db, { + ...options, + useTransactionClient: true, + }), + }; +} + +function requireScopeMember(member: T | undefined, name: string): T { + if (member !== undefined) return member; + throw new CompilerInvariantError( + `A PostgreSQL driver savepoint scope lost the ${name} execution member its transaction exposes.`, + { member: name }, + ); +} + +/** + * An adapter shaped like `outer` whose statements run on whichever scope + * `current` names when they execute. A transaction handle exposes no atomic + * batch, exclusive window or serialization classifier of its own, so these + * four I/O members are its complete surface. + */ +function routeThroughCurrentScope( + outer: SqlExecutionAdapter, + current: () => SqlExecutionAdapter, +): SqlExecutionAdapter { + return { + compile(query: ExecutableSql): CompiledSqlQuery { + return current().compile(query); + }, + execute(query: ExecutableSql): Promise { + return current().execute(query); + }, + ...(outer.executeCompiled === undefined ? + {} + : { + executeCompiled( + compiledQuery: CompiledSqlQuery, + ): Promise { + return requireScopeMember( + current().executeCompiled, + "executeCompiled", + )(compiledQuery); + }, + }), + ...(outer.prepare === undefined ? + {} + : { + prepare(sqlText: string): PreparedSqlStatement { + return { + execute( + params: readonly unknown[], + ): Promise { + return requireScopeMember(current().prepare, "prepare")( + sqlText, + ).execute(params); + }, + }; + }, + }), + }; +} + +export function createPostgresTransactionSession( + db: AnyPgDatabase, + options: PostgresExecutionAdapterOptions | undefined, +): PostgresTransactionSession { + const outer = bindScopeExecution(db, options); + if (!isPostgresJsClient(getPinnedPostgresTransactionClient(db))) { + return { + operation: outer.operation, + compiled: outer.compiled, + enterDriverScope: undefined, + }; + } + + let current = outer; + + // Drizzle's nested transaction on postgres-js is the driver's own + // `savepoint()` scope, bound to a Drizzle session over its handle. + async function enterDriverScope(scope: () => Promise): Promise { + const enclosing = current; + return enclosing.db.transaction(async (nested) => { + current = bindScopeExecution(nested, options); + try { + return await scope(); + } finally { + current = enclosing; + } + }); + } + + return { + operation: routeThroughCurrentScope( + outer.operation, + () => current.operation, + ), + compiled: routeThroughCurrentScope(outer.compiled, () => current.compiled), + enterDriverScope, + }; +} diff --git a/packages/typegraph/src/backend/drizzle/execution/session-atomic-batch.ts b/packages/typegraph/src/backend/drizzle/execution/session-atomic-batch.ts index 95d584f66..246d9c0ca 100644 --- a/packages/typegraph/src/backend/drizzle/execution/session-atomic-batch.ts +++ b/packages/typegraph/src/backend/drizzle/execution/session-atomic-batch.ts @@ -3,11 +3,17 @@ import type { AtomicSqlBatchExecutor, CompiledAtomicSqlStatement, } from "../../capabilities/atomic-sql-program"; +import { + type DriverSavepointScope, + runSessionSavepoint, + SavepointRecoveryFailure, +} from "../../session-savepoint"; import type { SqlExecutionAdapter } from "./types"; type SessionAtomicBatchExecution = Readonly<{ executeCompiled: NonNullable; runExclusive: NonNullable; + enterDriverScope?: DriverSavepointScope | undefined; }>; type SessionAtomicBatchAdapter = Readonly<{ @@ -15,18 +21,6 @@ type SessionAtomicBatchAdapter = Readonly<{ }>; const ATOMIC_PROGRAM_SAVEPOINT = "typegraph_atomic_program"; -const BEGIN_ATOMIC_PROGRAM = { - sql: `SAVEPOINT ${ATOMIC_PROGRAM_SAVEPOINT}`, - params: [], -} as const; -const ROLLBACK_ATOMIC_PROGRAM = { - sql: `ROLLBACK TO SAVEPOINT ${ATOMIC_PROGRAM_SAVEPOINT}`, - params: [], -} as const; -const RELEASE_ATOMIC_PROGRAM = { - sql: `RELEASE SAVEPOINT ${ATOMIC_PROGRAM_SAVEPOINT}`, - params: [], -} as const; /** * Adapts one already-open, exclusively reservable transaction session to the @@ -34,15 +28,20 @@ const RELEASE_ATOMIC_PROGRAM = { * * `runExclusive` is load-bearing: queueing each statement independently would * let unrelated work on the same transaction interleave with a program. The - * returned batch owns one queue slot for the whole sequence, while the outer - * transaction owns commit and rollback. A savepoint keeps deliberate + * returned batch owns one queue slot for the whole savepoint scope, while the + * outer transaction owns commit and rollback. A savepoint keeps deliberate * constraint refusals from aborting that outer transaction before Store * diagnostics can translate them to typed errors. */ export function createSessionAtomicBatchAdapter( execution: SessionAtomicBatchExecution, ): SessionAtomicBatchAdapter { - const { executeCompiled, runExclusive } = execution; + const { executeCompiled, runExclusive, enterDriverScope } = execution; + const port = { + executeControl: (statement: string) => + executeCompiled({ sql: statement, params: [] }), + enterDriverScope, + }; return { async executeAtomicBatch( @@ -50,31 +49,27 @@ export function createSessionAtomicBatchAdapter( ) { if (statements.length === 0) return []; return runExclusive(async () => { - await executeCompiled(BEGIN_ATOMIC_PROGRAM); - const results: (readonly TRow[])[] = []; try { - for (const statement of statements) { - results.push(await executeCompiled(statement)); - } - await executeCompiled(RELEASE_ATOMIC_PROGRAM); - return results; + // PostgreSQL protocol- and SQL-level prepared statements survive + // ROLLBACK TO SAVEPOINT (verified on PostgreSQL 18), so recovery + // restores data state without invalidating the session's cache. + return await runSessionSavepoint( + port, + ATOMIC_PROGRAM_SAVEPOINT, + async () => { + const results: (readonly TRow[])[] = []; + for (const statement of statements) { + results.push(await executeCompiled(statement)); + } + return { action: "release", value: results }; + }, + ); } catch (error) { - try { - // PostgreSQL protocol- and SQL-level prepared statements survive - // ROLLBACK TO SAVEPOINT (verified on PostgreSQL 18), so recovery - // restores data state without invalidating the session's cache. - await executeCompiled(ROLLBACK_ATOMIC_PROGRAM); - await executeCompiled(RELEASE_ATOMIC_PROGRAM); - } catch (recoveryError) { - const failures = new AggregateError( - [error, recoveryError], - "The atomic program failed and its savepoint recovery also failed.", - { cause: error }, - ); + if (error instanceof SavepointRecoveryFailure) { throw new CompilerInvariantError( "An atomic transaction session could not restore its savepoint after a failed program.", {}, - { cause: failures }, + { cause: error }, ); } throw error; diff --git a/packages/typegraph/src/backend/drizzle/postgres.ts b/packages/typegraph/src/backend/drizzle/postgres.ts index aea44d148..6137148bc 100644 --- a/packages/typegraph/src/backend/drizzle/postgres.ts +++ b/packages/typegraph/src/backend/drizzle/postgres.ts @@ -151,6 +151,10 @@ import { nowIso, POSTGRES_ROW_MAPPER_CONFIG, } from "../row-mappers"; +import { + registerDriverSavepointScope, + runSessionSavepoint, +} from "../session-savepoint"; import { resolveDeclaredBackendResource, type SerializedResourceDeclaration, @@ -250,6 +254,7 @@ import { type PostgresExecutionAdapter, type PostgresExecutionAdapterOptions, } from "./execution/postgres-execution"; +import { createPostgresTransactionSession } from "./execution/postgres-transaction-session"; import { createSessionAtomicBatchAdapter } from "./execution/session-atomic-batch"; import { createSerialExecutionAdapter } from "./execution/statement-queue"; import { type ExecutableSql, toDrizzleSql } from "./execution/types"; @@ -2456,12 +2461,17 @@ function buildPostgresEngineProfileInternal( // SAVEPOINT refuses outside an explicit transaction. Release it // immediately: this proves the literal pinned session is active, // without opening another transaction or changing application data. + // The scope issues nothing, so no driver scope is needed to + // attribute a failure inside it. try { - await externalTx.execute( - sql`SAVEPOINT typegraph_schema_adoption_probe`, - ); - await externalTx.execute( - sql`RELEASE SAVEPOINT typegraph_schema_adoption_probe`, + await runSessionSavepoint( + { + executeControl: (statement) => + externalTx.execute(sql.raw(statement)), + }, + "typegraph_schema_adoption_probe", + () => + Promise.resolve({ action: "release", value: undefined }), ); } catch (error) { throw new ConfigurationError( @@ -4000,28 +4010,29 @@ function createTransactionBackend( // The wrap is unconditional because the overlap is created inside TypeGraph // (the node write pipeline syncs embeddings and fulltext concurrently), not // only by user code. - const operationExecutionAdapter = createPostgresExecutionAdapter( + const session = createPostgresTransactionSession( options.db, options.adapterOptions, ); - const compiledExecutionAdapter = createPostgresExecutionAdapter(options.db, { - ...options.adapterOptions, - useTransactionClient: true, - }); - const executeCompiled = compiledExecutionAdapter.executeCompiled; + const executeCompiled = session.compiled.executeCompiled; + const { enterDriverScope } = session; // Preserve the established Drizzle-routed execution surface for ordinary // transaction work. Only a closed atomic program needs the pinned client's // compiled-text seam; changing every statement to that seam would also // bypass driver logging and other Drizzle session behavior merely because // this transaction can execute programs. const txExecutionAdapter = createSerialExecutionAdapter( - operationExecutionAdapter, + session.operation, getPinnedPostgresTransactionClient(options.db) ?? options.db, ); const runExclusive = txExecutionAdapter.runExclusive; const sessionAtomicBatchAdapter = executeCompiled === undefined ? undefined : ( - createSessionAtomicBatchAdapter({ executeCompiled, runExclusive }) + createSessionAtomicBatchAdapter({ + executeCompiled, + runExclusive, + enterDriverScope, + }) ); const sessionAtomicSqlProgramExecutor = sessionAtomicBatchAdapter === undefined ? undefined : ( @@ -4089,6 +4100,9 @@ function createTransactionBackend( options.isFirstParty ? markFirstPartyFactory(txOperationBackend) : txOperationBackend; + if (enterDriverScope !== undefined) { + registerDriverSavepointScope(backend, enterDriverScope); + } if (sessionAtomicBatchAdapter !== undefined) { registerAtomicSqlProgram(backend, sessionAtomicBatchAdapter); diff --git a/packages/typegraph/src/backend/session-savepoint.ts b/packages/typegraph/src/backend/session-savepoint.ts new file mode 100644 index 000000000..d6d009f5b --- /dev/null +++ b/packages/typegraph/src/backend/session-savepoint.ts @@ -0,0 +1,154 @@ +/** + * The one savepoint primitive for work TypeGraph recovers inside an open + * transaction session. + * + * TypeGraph isolates a statement that may fail on purpose — a fused write + * program's deliberate abort, an imported row the database refuses — behind a + * savepoint, rolls back to it, and reports the typed refusal while the + * caller's transaction stays usable. The savepoint itself is plain SQL issued + * through the session's own control channel. + * + * Some drivers additionally track the first failed statement of each + * transaction scope and rethrow it when that scope ends, unless the statement + * ran inside one of the driver's own savepoint scopes (postgres-js does). A + * session on such a driver supplies a {@link DriverSavepointScope}; the + * primitive enters it between its own `SAVEPOINT` and `RELEASE`, so a failure + * inside the scope is attributed to the nested driver scope and the enclosing + * one never sees it. The driver scope sits inside TypeGraph's own savepoint, + * whose release also releases the driver's — a driver that never releases its + * scopes would otherwise nest every recovered savepoint one level deeper for + * the rest of the transaction. + */ +import { backendDerivationChain } from "./derive-backend"; + +/** What a savepoint scope decided once its work ran without throwing. */ +export type SavepointDecision = + | Readonly<{ action: "release"; value: T }> + | Readonly<{ action: "rollback"; value: T; cause: unknown }>; + +/** + * Runs `scope` inside one of the driver's own savepoint scopes. A throw rolls + * the driver scope back and propagates; a resolution leaves it to the + * enclosing TypeGraph savepoint. + */ +export type DriverSavepointScope = (scope: () => Promise) => Promise; + +type SessionSavepointPort = Readonly<{ + /** Issues one savepoint control statement on the enclosing session handle. */ + executeControl: (statement: string) => Promise; + enterDriverScope?: DriverSavepointScope | undefined; +}>; + +/** The scope failed and rolling its savepoint back failed too. */ +export class SavepointRecoveryFailure extends AggregateError { + constructor(cause: unknown, recoveryError: unknown) { + super( + [cause, recoveryError], + "A TypeGraph savepoint could not be restored after its scope failed.", + { cause }, + ); + this.name = "SavepointRecoveryFailure"; + } +} + +/** Carries a rollback decision out of a driver scope, which rolls back only on a throw. */ +class SavepointRollbackDecision extends Error { + constructor(readonly decision: SavepointDecision) { + super("TypeGraph savepoint rollback decision"); + this.name = "SavepointRollbackDecision"; + } +} + +async function decideInsideDriverScope( + enterDriverScope: DriverSavepointScope | undefined, + scope: () => Promise>, +): Promise> { + if (enterDriverScope === undefined) return scope(); + try { + return await enterDriverScope(async () => { + const decision = await scope(); + if (decision.action === "rollback") { + throw new SavepointRollbackDecision(decision); + } + return decision; + }); + } catch (error) { + if (error instanceof SavepointRollbackDecision) { + return error.decision; + } + throw error; + } +} + +/** + * Runs `scope` inside savepoint `savepoint` on the session behind `port`. + * + * A `release` decision releases the savepoint and returns its value. A + * `rollback` decision, or a throw, rolls back to the savepoint, calls + * `onRolledBack`, and releases it; the decision's value is returned, the + * thrown error rethrown. A failed rollback is reported as a + * {@link SavepointRecoveryFailure}, since the session can no longer be + * trusted to be at the savepoint's state. + */ +export async function runSessionSavepoint( + port: SessionSavepointPort, + savepoint: string, + scope: () => Promise>, + onRolledBack?: () => void, +): Promise { + const { executeControl, enterDriverScope } = port; + + async function rollBack(cause: unknown): Promise { + try { + await executeControl(`ROLLBACK TO SAVEPOINT ${savepoint}`); + onRolledBack?.(); + await executeControl(`RELEASE SAVEPOINT ${savepoint}`); + } catch (recoveryError) { + throw new SavepointRecoveryFailure(cause, recoveryError); + } + } + + await executeControl(`SAVEPOINT ${savepoint}`); + let decision: SavepointDecision; + try { + decision = await decideInsideDriverScope(enterDriverScope, scope); + if (decision.action === "release") { + await executeControl(`RELEASE SAVEPOINT ${savepoint}`); + return decision.value; + } + } catch (error) { + await rollBack(error); + throw error; + } + await rollBack(decision.cause); + return decision.value; +} + +const DRIVER_SAVEPOINT_SCOPES = new WeakMap(); + +/** + * Registers the driver scope of one transaction session's backend. Only a + * session whose driver attributes failed statements to their own scope + * registers one. + */ +export function registerDriverSavepointScope( + backend: object, + enterDriverScope: DriverSavepointScope, +): void { + DRIVER_SAVEPOINT_SCOPES.set(backend, enterDriverScope); +} + +/** + * The driver scope of the session `target` executes on. Backends derived from + * a session's backend issue their statements on that same session, so they + * resolve its scope along their derivation chain. + */ +export function resolveDriverSavepointScope( + target: object, +): DriverSavepointScope | undefined { + for (const link of backendDerivationChain(target)) { + const enterDriverScope = DRIVER_SAVEPOINT_SCOPES.get(link); + if (enterDriverScope !== undefined) return enterDriverScope; + } + return undefined; +} diff --git a/packages/typegraph/src/store/recorded-capture.ts b/packages/typegraph/src/store/recorded-capture.ts index 4469ce639..4280372a0 100644 --- a/packages/typegraph/src/store/recorded-capture.ts +++ b/packages/typegraph/src/store/recorded-capture.ts @@ -17,6 +17,11 @@ import { isBackendDerivedFrom, projectGraphBackend, } from "../backend/derive-backend"; +import { + resolveDriverSavepointScope, + runSessionSavepoint, + type SavepointDecision, +} from "../backend/session-savepoint"; import { type DeleteEdgesBatchParams, type EdgeRow, @@ -495,10 +500,6 @@ type TransactionControlTarget = Readonly< Pick >; -type RecordedSavepointDecision = - | Readonly<{ action: "release"; value: T }> - | Readonly<{ action: "rollback"; value: T; cause: unknown }>; - const recordedTransactionBindings = new WeakMap< object, RecordedTransactionBinding @@ -550,7 +551,7 @@ export async function runRecordedTransactionSavepoint( { supported: true } >, savepoint: string, - fn: () => Promise>, + fn: () => Promise>, ): Promise { const binding = recordedTransactionBindings.get(target); binding?.assertOpen(); @@ -581,36 +582,17 @@ export async function runRecordedTransactionSavepoint( binding.capture.graphLocks.set(graphId, lock); } }; - const executeControl = (sql: string): Promise => - executeStatement(asCompiledStatementSql(portableSql.raw(sql))); - const rollback = async (cause: unknown): Promise => { - try { - await executeControl(`ROLLBACK TO SAVEPOINT ${savepoint}`); - restoreCapture(); - await executeControl(`RELEASE SAVEPOINT ${savepoint}`); - } catch (recoveryError) { - throw new AggregateError( - [cause], - "Failed to recover a TypeGraph recorded transaction savepoint.", - { cause: recoveryError }, - ); - } - }; - await executeControl(`SAVEPOINT ${savepoint}`); - let result: RecordedSavepointDecision; - try { - result = await fn(); - } catch (error) { - await rollback(error); - throw error; - } - if (result.action === "rollback") { - await rollback(result.cause); - } else { - await executeControl(`RELEASE SAVEPOINT ${savepoint}`); - } - return result.value; + return runSessionSavepoint( + { + executeControl: (statement) => + executeStatement(asCompiledStatementSql(portableSql.raw(statement))), + enterDriverScope: resolveDriverSavepointScope(rawTarget), + }, + savepoint, + fn, + restoreCapture, + ); } /** Internal fence/control statements bypass row capture but retain its lifetime guard. */ diff --git a/packages/typegraph/tests/backends/integration-test-suite.ts b/packages/typegraph/tests/backends/integration-test-suite.ts index 9ac591acc..9a79b39bc 100644 --- a/packages/typegraph/tests/backends/integration-test-suite.ts +++ b/packages/typegraph/tests/backends/integration-test-suite.ts @@ -96,6 +96,7 @@ import { registerRecursiveStopExpansionIntegrationTests, registerRelationalCompositionIntegrationTests, registerRemovalMaterializationIntegrationTests, + registerSavepointRecoveryIntegrationTests, registerSchemaCheckedReadIntegrationTests, registerSelectiveRetryIntegrationTests, registerSetNodeMutationIntegrationTests, @@ -340,6 +341,7 @@ export function createIntegrationTestSuite< registerClaimLookupPreferenceIntegrationTests(context); registerConstraintFenceErrorIntegrationTests(context); registerConstraintFenceTransactionHealthTests(context); + registerSavepointRecoveryIntegrationTests(context); registerConstraintFenceVerificationIntegrationTests(context); registerRecordedReadBindingIntegrationTests(context); registerSetOperationIntegrationTests(context); diff --git a/packages/typegraph/tests/backends/integration/adopted-evolution.ts b/packages/typegraph/tests/backends/integration/adopted-evolution.ts index 6bc533905..c8121903b 100644 --- a/packages/typegraph/tests/backends/integration/adopted-evolution.ts +++ b/packages/typegraph/tests/backends/integration/adopted-evolution.ts @@ -184,6 +184,37 @@ export function registerAdoptedEvolutionIntegrationTests( expect(active?.version).toBe(plan.result.version); }); + it("refuses to adopt a native transaction handle whose transaction has ended", async () => { + const backend = context.getBackend(); + const [store] = await createAdapterStoreWithSchema(graph, backend); + const adopt = backend.adoptSchemaWriteTransaction; + if (adopt === undefined) { + const plan = await store.planEvolution(extension); + await assertUnsupportedAdoption(() => + backend.transactionWithNative(async (_target, nativeTx) => + store.withEvolvedTransaction(nativeTx, plan, async () => { + await Promise.resolve(); + }), + ), + ); + return; + } + const endedTransaction = await backend.transactionWithNative( + (_target, nativeTx) => Promise.resolve(nativeTx), + ); + + await expect( + adopt(endedTransaction, graph.id, { waitBudgetMs: 1000 }), + ).rejects.toBeInstanceOf(ConfigurationError); + + const person = await store.nodes.AdoptedPerson.create({ + name: "after the refused adoption", + }); + await expect( + store.nodes.AdoptedPerson.getById(person.id), + ).resolves.toMatchObject({ name: "after the refused adoption" }); + }); + it("refuses copied plans and plans for another graph before the callback", async () => { const backend = context.getBackend(); const [store] = await createAdapterStoreWithSchema(graph, backend); diff --git a/packages/typegraph/tests/backends/integration/index.ts b/packages/typegraph/tests/backends/integration/index.ts index a5cd30b49..7a3df849e 100644 --- a/packages/typegraph/tests/backends/integration/index.ts +++ b/packages/typegraph/tests/backends/integration/index.ts @@ -68,6 +68,7 @@ export { registerRecursiveIntegrationTests } from "./recursive"; export { registerRecursiveStopExpansionIntegrationTests } from "./recursive-stop-expansion"; export { registerRelationalCompositionIntegrationTests } from "./relational-composition"; export { registerRemovalMaterializationIntegrationTests } from "./removal-materialization"; +export { registerSavepointRecoveryIntegrationTests } from "./savepoint-recovery"; export { registerSchemaCheckedReadIntegrationTests } from "./schema-checked-read"; export { registerSelectiveRetryIntegrationTests } from "./selective-retry"; export { registerSetNodeMutationIntegrationTests } from "./set-node-mutation"; diff --git a/packages/typegraph/tests/backends/integration/savepoint-recovery.ts b/packages/typegraph/tests/backends/integration/savepoint-recovery.ts new file mode 100644 index 000000000..9cfc2d484 --- /dev/null +++ b/packages/typegraph/tests/backends/integration/savepoint-recovery.ts @@ -0,0 +1,230 @@ +/** + * A refusal TypeGraph recovers through a savepoint leaves the enclosing + * transaction committable. + * + * Some writes can only learn that they lost from the database itself: a fused + * write program whose deliberate aborting statement reports a lost unique key, + * or an imported edge whose durable match identity is arbitrated by a unique + * index. On PostgreSQL that failed statement aborts the transaction, so + * TypeGraph isolates it behind a savepoint, rolls back to it, and raises the + * typed refusal. The transaction must then commit everything written outside + * the savepoint, on every driver — including drivers that track the first + * failed statement of a transaction and rethrow it at commit unless it was + * issued inside one of their own savepoint scopes. + */ +import { describe, expect, it } from "vitest"; +import { z } from "zod"; + +import { + asEdgeId, + createStoreWithSchema, + defineEdge, + defineGraph, + defineNode, + UniquenessError, +} from "../../../src"; +import { + deriveBackend, + projectBackendWithout, + projectGraphBackend, +} from "../../../src/backend/derive-backend"; +import { type GraphBackend } from "../../../src/backend/types"; +import { type GraphData, importGraph } from "../../../src/interchange"; +import { type IntegrationTestContext } from "./test-context"; + +const SavepointStamp = defineNode("SavepointStamp", { + schema: z.object({ label: z.string() }), +}); +/** Unconstrained, so writing one asserts transaction health only. */ +const SavepointNote = defineNode("SavepointNote", { + schema: z.object({ body: z.string() }), +}); + +const stampGraph = defineGraph({ + id: "savepoint_recovery_stamps", + nodes: { + SavepointStamp: { + type: SavepointStamp, + unique: [ + { + name: "savepoint_stamp_label", + fields: ["label"], + scope: "kind", + collation: "binary", + }, + ], + }, + SavepointNote: { type: SavepointNote }, + }, + edges: {}, +}); + +const SavepointPerson = defineNode("SavepointPerson", { + schema: z.object({ name: z.string() }), +}); +const savepointKnows = defineEdge("savepointKnows", { + schema: z.object({ label: z.string() }), +}); + +const identityGraph = defineGraph({ + id: "savepoint_recovery_edge_identity", + nodes: { SavepointPerson: { type: SavepointPerson } }, + edges: { + savepointKnows: { + type: savepointKnows, + from: [SavepointPerson], + to: [SavepointPerson], + matchIdentity: { name: "savepoint-knows-label", fields: ["label"] }, + }, + }, +}); + +/** + * Hides every application-side owner read of a durable edge identity, so the + * database's unique index is what refuses a duplicate and the import has to + * recover that failed statement through its per-row savepoint. + */ +function backendWithBlindEdgeIdentityProbes( + backend: GraphBackend, +): GraphBackend { + return deriveBackend( + projectBackendWithout(projectGraphBackend(backend), [ + "insertEdgesDurableBatchReturning", + ]), + { + findEdgesByHeterogeneousEndpointSet: () => Promise.resolve([]), + transaction: (run, options) => + backend.transaction( + (target) => + run( + deriveBackend( + projectBackendWithout(target, [ + "insertEdgesDurableBatchReturning", + ]), + { + findEdgesByHeterogeneousEndpointSet: () => + Promise.resolve([]), + }, + ), + ), + options, + ), + }, + ); +} + +function edgeIdentityDocument(edges: GraphData["edges"]): GraphData { + return { + formatVersion: "2.0", + exportedAt: "2026-01-01T00:00:00.000Z", + source: { + type: "typegraph-export", + graphId: identityGraph.id, + schemaVersion: 1, + }, + nodes: [], + edges, + }; +} + +function knowsEdge(id: string, label: string): GraphData["edges"][number] { + return { + kind: "savepointKnows", + id, + from: { kind: "SavepointPerson", id: "owner" }, + to: { kind: "SavepointPerson", id: "target" }, + properties: { label }, + }; +} + +export function registerSavepointRecoveryIntegrationTests( + context: IntegrationTestContext, +): void { + describe("a refusal recovered through a savepoint", () => { + it("lets a caller transaction commit after bulkReplaceById loses a unique key", async () => { + const store = await context.createStore(stampGraph); + const first = await store.nodes.SavepointStamp.create({ label: "one" }); + const second = await store.nodes.SavepointStamp.create({ label: "two" }); + await store.nodes.SavepointStamp.create({ label: "taken" }); + + await store.transaction(async (tx) => { + await tx.nodes.SavepointNote.create( + { body: "written before the refusal" }, + { id: "savepoint-note-before" }, + ); + await expect( + tx.nodes.SavepointStamp.bulkReplaceById([ + { id: second.id, props: { label: "taken" } }, + { id: first.id, props: { label: "renamed" } }, + ]), + ).rejects.toThrow(UniquenessError); + await tx.nodes.SavepointNote.create( + { body: "written after the refusal" }, + { id: "savepoint-note-after" }, + ); + }); + + const notes = await store.nodes.SavepointNote.find(); + expect(notes.map((note) => note.body).toSorted()).toEqual([ + "written after the refusal", + "written before the refusal", + ]); + const stamps = await store.nodes.SavepointStamp.find(); + expect(stamps.map((stamp) => stamp.label).toSorted()).toEqual([ + "one", + "taken", + "two", + ]); + }); + + it("commits a history-capturing import whose deferred edge row loses its durable identity", async () => { + const [setupStore] = await createStoreWithSchema( + identityGraph, + context.getBackend(), + ); + const owner = await setupStore.nodes.SavepointPerson.create( + { name: "Owner" }, + { id: "owner" }, + ); + const target = await setupStore.nodes.SavepointPerson.create( + { name: "Target" }, + { id: "target" }, + ); + await setupStore.edges.savepointKnows.create(owner, target, { + label: "taken", + }); + + const [historyStore] = await createStoreWithSchema( + identityGraph, + backendWithBlindEdgeIdentityProbes(context.getBackend()), + { history: true }, + ); + // Two rows sharing an id are deferred to the per-row path, where each + // create runs inside its own savepoint. The first is refused for its + // oversized identity key; the second reaches the database, whose unique + // index refuses its already-owned label inside that savepoint. + const result = await importGraph( + historyStore, + edgeIdentityDocument([ + knowsEdge("savepoint-unrelated", "other"), + knowsEdge("savepoint-conflict", "x".repeat(3000)), + knowsEdge("savepoint-conflict", "taken"), + ]), + { onConflict: "error", batchSize: 100, refreshStatistics: false }, + ); + + expect(result.edges.created).toBe(1); + expect(result.errors).toHaveLength(2); + await expect( + historyStore.edges.savepointKnows.getById( + asEdgeId("savepoint-unrelated"), + ), + ).resolves.toMatchObject({ label: "other" }); + await expect( + historyStore.edges.savepointKnows.getById( + asEdgeId("savepoint-conflict"), + ), + ).resolves.toBeUndefined(); + }); + }); +} diff --git a/packages/typegraph/tests/recorded-capture-write-parity.test.ts b/packages/typegraph/tests/recorded-capture-write-parity.test.ts index 7a91d4b1a..9a632c4d6 100644 --- a/packages/typegraph/tests/recorded-capture-write-parity.test.ts +++ b/packages/typegraph/tests/recorded-capture-write-parity.test.ts @@ -6,6 +6,7 @@ import { } from "../src/backend/capabilities/resolve"; import { deriveBackend } from "../src/backend/derive-backend"; import { createLocalPgliteBackend } from "../src/backend/postgres/pglite"; +import { SavepointRecoveryFailure } from "../src/backend/session-savepoint"; import { createSqlSchema } from "../src/query/compiler/schema"; import { createRecordedBackend, @@ -202,7 +203,7 @@ describe("recorded-capture write-surface parity", () => { }; }, ), - ).rejects.toThrow("Failed to recover a TypeGraph recorded transaction"); + ).rejects.toBeInstanceOf(SavepointRecoveryFailure); expect(await scope.flush()).toEqual(new Map()); }); }); diff --git a/packages/typegraph/tests/session-savepoint.test.ts b/packages/typegraph/tests/session-savepoint.test.ts new file mode 100644 index 000000000..026da5d29 --- /dev/null +++ b/packages/typegraph/tests/session-savepoint.test.ts @@ -0,0 +1,171 @@ +import { describe, expect, it } from "vitest"; + +import { deriveBackend } from "../src/backend/derive-backend"; +import { + type DriverSavepointScope, + registerDriverSavepointScope, + resolveDriverSavepointScope, + runSessionSavepoint, + SavepointRecoveryFailure, +} from "../src/backend/session-savepoint"; +import { createTestBackend } from "./test-utils"; + +/** + * A session whose control statements and driver-scope boundaries land in one + * ordered log, so each case can assert exactly where the driver scope sits + * relative to TypeGraph's own savepoint. + */ +function createRecordingSession( + options: Readonly<{ failingControl?: string }> = {}, +) { + const events: string[] = []; + const enterDriverScope: DriverSavepointScope = async (scope) => { + events.push("driver scope opened"); + try { + const value = await scope(); + events.push("driver scope left open"); + return value; + } catch (error) { + events.push("driver scope rolled back"); + throw error; + } + }; + return { + events, + port: { + executeControl: (statement: string) => { + events.push(statement); + return statement.startsWith(options.failingControl ?? "\0") ? + Promise.reject(new Error(`injected ${statement} failure`)) + : Promise.resolve(); + }, + enterDriverScope, + }, + }; +} + +const passThroughDriverScope: DriverSavepointScope = (scope) => scope(); + +describe("session savepoints", () => { + it("opens the driver scope inside the savepoint whose release also releases it", async () => { + const { events, port } = createRecordingSession(); + + const value = await runSessionSavepoint(port, "sp_release", () => { + events.push("scope work"); + return Promise.resolve({ action: "release", value: "released" }); + }); + + expect(value).toBe("released"); + expect(events).toEqual([ + "SAVEPOINT sp_release", + "driver scope opened", + "scope work", + "driver scope left open", + "RELEASE SAVEPOINT sp_release", + ]); + }); + + it("rolls a driver scope back for a rollback decision and still returns its value", async () => { + const { events, port } = createRecordingSession(); + let restored = 0; + + const value = await runSessionSavepoint( + port, + "sp_decision", + () => + Promise.resolve({ + action: "rollback", + value: "refused row", + cause: new Error("refusal"), + }), + () => { + restored += 1; + events.push("restored"); + }, + ); + + expect(value).toBe("refused row"); + expect(restored).toBe(1); + expect(events).toEqual([ + "SAVEPOINT sp_decision", + "driver scope opened", + "driver scope rolled back", + "ROLLBACK TO SAVEPOINT sp_decision", + "restored", + "RELEASE SAVEPOINT sp_decision", + ]); + }); + + it("rolls back and rethrows the scope's own failure", async () => { + const { events, port } = createRecordingSession(); + const failure = new Error("statement failed"); + + await expect( + runSessionSavepoint(port, "sp_throw", () => Promise.reject(failure)), + ).rejects.toBe(failure); + expect(events).toEqual([ + "SAVEPOINT sp_throw", + "driver scope opened", + "driver scope rolled back", + "ROLLBACK TO SAVEPOINT sp_throw", + "RELEASE SAVEPOINT sp_throw", + ]); + }); + + it("recovers a failed release by rolling the savepoint back", async () => { + const releaseFailure = new Error("release failed"); + const events: string[] = []; + let releases = 0; + + await expect( + runSessionSavepoint( + { + executeControl: (statement) => { + events.push(statement); + if (statement.startsWith("RELEASE") && releases++ === 0) { + return Promise.reject(releaseFailure); + } + return Promise.resolve(); + }, + }, + "sp_release_failure", + () => Promise.resolve({ action: "release", value: undefined }), + ), + ).rejects.toBe(releaseFailure); + expect(events).toEqual([ + "SAVEPOINT sp_release_failure", + "RELEASE SAVEPOINT sp_release_failure", + "ROLLBACK TO SAVEPOINT sp_release_failure", + "RELEASE SAVEPOINT sp_release_failure", + ]); + }); + + it("reports a failed rollback with both failures", async () => { + const { port } = createRecordingSession({ failingControl: "ROLLBACK TO" }); + const failure = new Error("statement failed"); + + const error = await runSessionSavepoint(port, "sp_broken", () => + Promise.reject(failure), + ).catch((error_: unknown) => error_); + + expect(error).toBeInstanceOf(SavepointRecoveryFailure); + expect((error as SavepointRecoveryFailure).cause).toBe(failure); + expect((error as SavepointRecoveryFailure).errors).toEqual([ + failure, + new Error("injected ROLLBACK TO SAVEPOINT sp_broken failure"), + ]); + }); + + it("resolves a session's driver scope through backends derived from it", () => { + const sessionBackend = createTestBackend(); + const unrelatedBackend = createTestBackend(); + registerDriverSavepointScope(sessionBackend, passThroughDriverScope); + + const derivedBackend = deriveBackend(deriveBackend(sessionBackend, {}), {}); + + expect(resolveDriverSavepointScope(derivedBackend)).toBe( + passThroughDriverScope, + ); + expect(resolveDriverSavepointScope(unrelatedBackend)).toBeUndefined(); + }); +}); From f02a48aeaf1c099e68ed35ee8841a14bc2a3f9f8 Mon Sep 17 00:00:00 2001 From: Paul Dlug Date: Fri, 23 Oct 2026 14:08:20 -0700 Subject: [PATCH 2/2] fix(postgres): refuse overlapping driver savepoint scopes on one transaction The postgres-js transaction session kept its open driver scope in one slot and restored the enclosing scope on exit, which is correct only while scopes nest strictly. Two savepoint-protected writes running concurrently on one transaction can overlap without nesting; when the earlier scope exited first, statements of the still-open scope were routed outside it, and its own exit then restored a scope that had already ended. Open scopes are now a stack. An exit that is not the innermost open scope throws CompilerInvariantError (code OVERLAPPING_SAVEPOINT_SCOPES), and once an ended scope is uncovered every later statement on the session refuses with the same error instead of running on an ended scope or the outer handle. --- .../execution/postgres-transaction-session.ts | 56 +++++++-- .../postgres-transaction-session.test.ts | 115 ++++++++++++++++++ 2 files changed, 162 insertions(+), 9 deletions(-) create mode 100644 packages/typegraph/tests/postgres-transaction-session.test.ts diff --git a/packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts b/packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts index 8ad29d297..bf3dfe3b7 100644 --- a/packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts +++ b/packages/typegraph/src/backend/drizzle/execution/postgres-transaction-session.ts @@ -31,6 +31,13 @@ type ScopeExecution = Readonly<{ * surfaces issue every statement through its handle: PostgreSQL places them * inside the savepoint whichever handle sent them, and routing them through it * keeps the driver's attribution in step with the server. + * + * Driver scopes on one session must nest strictly: the scope that exits is + * always the innermost one still open. Two scopes that overlap without nesting + * (both opened by concurrent work in one transaction, the outer one exiting + * first) have no single innermost scope to route to, so the out-of-order exit + * throws a `CompilerInvariantError`, and every later statement on the session + * refuses with the same error instead of running on a scope that has ended. */ type PostgresTransactionSession = Readonly<{ operation: SqlExecutionAdapter; @@ -66,6 +73,14 @@ function requireScopeMember(member: T | undefined, name: string): T { * batch, exclusive window or serialization classifier of its own, so these * four I/O members are its complete surface. */ +function overlappingScopes(cause?: unknown): CompilerInvariantError { + return new CompilerInvariantError( + "Overlapping savepoint scopes in one PostgreSQL transaction: a driver savepoint scope ended while a scope opened after it was still open. Savepoint-protected writes on one transaction must not run concurrently.", + { code: "OVERLAPPING_SAVEPOINT_SCOPES" }, + cause === undefined ? undefined : { cause }, + ); +} + function routeThroughCurrentScope( outer: SqlExecutionAdapter, current: () => SqlExecutionAdapter, @@ -120,28 +135,51 @@ export function createPostgresTransactionSession( }; } - let current = outer; + const openScopes: ScopeExecution[] = []; + const endedScopes = new WeakSet(); + + function current(): ScopeExecution { + const innermost = openScopes.at(-1); + if (innermost === undefined) return outer; + if (endedScopes.has(innermost)) throw overlappingScopes(); + return innermost; + } + + function leave(scope: ScopeExecution, cause?: unknown): void { + endedScopes.add(scope); + if (openScopes.at(-1) !== scope) { + throw overlappingScopes(cause); + } + openScopes.pop(); + } // Drizzle's nested transaction on postgres-js is the driver's own // `savepoint()` scope, bound to a Drizzle session over its handle. async function enterDriverScope(scope: () => Promise): Promise { - const enclosing = current; - return enclosing.db.transaction(async (nested) => { - current = bindScopeExecution(nested, options); + return current().db.transaction(async (nested) => { + const entered = bindScopeExecution(nested, options); + openScopes.push(entered); + let result: T; try { - return await scope(); - } finally { - current = enclosing; + result = await scope(); + } catch (error) { + leave(entered, error); + throw error; } + leave(entered); + return result; }); } return { operation: routeThroughCurrentScope( outer.operation, - () => current.operation, + () => current().operation, + ), + compiled: routeThroughCurrentScope( + outer.compiled, + () => current().compiled, ), - compiled: routeThroughCurrentScope(outer.compiled, () => current.compiled), enterDriverScope, }; } diff --git a/packages/typegraph/tests/postgres-transaction-session.test.ts b/packages/typegraph/tests/postgres-transaction-session.test.ts new file mode 100644 index 000000000..bbbf25a66 --- /dev/null +++ b/packages/typegraph/tests/postgres-transaction-session.test.ts @@ -0,0 +1,115 @@ +import { describe, expect, it } from "vitest"; + +import { type AnyPgDatabase } from "../src/backend/drizzle/execution/postgres-execution"; +import { createPostgresTransactionSession } from "../src/backend/drizzle/execution/postgres-transaction-session"; +import { CompilerInvariantError } from "../src/errors"; +import { requireDefined } from "../src/utils/presence"; + +/** + * A Drizzle transaction over a postgres-js-shaped client. Every statement the + * pinned client runs is logged with the name of the scope that ran it, and a + * nested `transaction()` hands out a child scope named after its parent. + */ +function fakePostgresJsTransaction(name: string, log: string[]): AnyPgDatabase { + let children = 0; + const client = Object.assign((): undefined => undefined, { + unsafe: (sqlText: string): Promise => { + log.push(`${name}: ${sqlText}`); + return Promise.resolve([]); + }, + savepoint: (): never => { + throw new Error("The session must open scopes through Drizzle."); + }, + }); + const transaction = { + session: { client }, + execute: (): Promise => Promise.resolve([]), + transaction: ( + callback: (nested: AnyPgDatabase) => Promise, + ): Promise => { + children += 1; + return callback(fakePostgresJsTransaction(`${name}.${children}`, log)); + }, + }; + return transaction as unknown as AnyPgDatabase; +} + +function deferred(): Readonly<{ promise: Promise; resolve: () => void }> { + let resolve: (() => void) | undefined; + const promise = new Promise((settle) => { + resolve = settle; + }); + return { promise, resolve: requireDefined(resolve) }; +} + +function createRecordedSession() { + const log: string[] = []; + const session = createPostgresTransactionSession( + fakePostgresJsTransaction("outer", log), + undefined, + ); + const executeCompiled = requireDefined(session.compiled.executeCompiled); + return { + log, + run: async (sqlText: string): Promise => { + await executeCompiled({ sql: sqlText, params: [] }); + }, + enterDriverScope: requireDefined(session.enterDriverScope), + }; +} + +describe("PostgreSQL transaction session driver scopes", () => { + it("routes statements through the innermost open driver scope and back out", async () => { + const { log, run, enterDriverScope } = createRecordedSession(); + + await run("before"); + await enterDriverScope(async () => { + await run("in scope"); + await enterDriverScope(() => run("in nested scope")); + await run("back in scope"); + }); + await run("after"); + + expect(log).toEqual([ + "outer: before", + "outer.1: in scope", + "outer.1.1: in nested scope", + "outer.1: back in scope", + "outer: after", + ]); + }); + + it("refuses overlapping driver scopes instead of routing statements outside or into an ended scope", async () => { + const { log, run, enterDriverScope } = createRecordedSession(); + const firstGate = deferred(); + const secondGate = deferred(); + + const first = enterDriverScope(() => firstGate.promise); + const second = enterDriverScope(async () => { + await run("in second"); + await secondGate.promise; + }); + + firstGate.resolve(); + const firstFailure = await first.catch((error: unknown) => error); + expect(firstFailure).toBeInstanceOf(CompilerInvariantError); + expect(firstFailure).toMatchObject({ + details: { code: "OVERLAPPING_SAVEPOINT_SCOPES" }, + }); + + // The scope opened second is still open, so it keeps its own statements. + await run("while second is open"); + secondGate.resolve(); + await second; + + // Its exit uncovers the ended first scope: nothing may run there or leak + // to the outer handle. + await expect(run("after both")).rejects.toMatchObject({ + details: { code: "OVERLAPPING_SAVEPOINT_SCOPES" }, + }); + expect(log).toEqual([ + "outer.1.1: in second", + "outer.1.1: while second is open", + ]); + }); +});