diff --git a/apps/cli/docs/go-cli-divergences.md b/apps/cli/docs/go-cli-divergences.md index 228fc2af93..ee69236046 100644 --- a/apps/cli/docs/go-cli-divergences.md +++ b/apps/cli/docs/go-cli-divergences.md @@ -230,3 +230,11 @@ These commands exist in the TS CLI today but have no direct top-level equivalent pull."). An in-sync database is a finding, not a failure to troubleshoot, so the debug hint sent users chasing a non-existent bug. Message text and exit code — the parts scripts depend on — are unchanged. +- A migration batch that never reached the wire (the connection died before or during + submit) fails as a connection error carrying the driver's reason, with no + `At statement: N` line and no statement echo. Go's `formatError` + (`apps/cli-go/pkg/migration/file.go:126-147`) renders every `ExecBatch` failure — + dead connection included — as `\nAt statement: N\n`. Naming a statement + that provably never ran sent users debugging their own SQL for a transport failure, + so the TS shell reports the connectivity failure instead. A batch that was written + keeps Go's rendering unchanged. diff --git a/apps/cli/src/legacy/shared/legacy-db-connection.service.ts b/apps/cli/src/legacy/shared/legacy-db-connection.service.ts index 674876e74c..e19ce440b8 100644 --- a/apps/cli/src/legacy/shared/legacy-db-connection.service.ts +++ b/apps/cli/src/legacy/shared/legacy-db-connection.service.ts @@ -113,9 +113,10 @@ export interface LegacyDbSession { * statements that completed before the error. * * A batch runs on its own pooled connection, which the driver checks out per - * call. Failing to acquire it raises `LegacyDbConnectError` (a connection-setup - * failure, surfaced verbatim — not masked as an exec error), consistent with - * {@link queryRaw}; only the batch's own execution raises `LegacyDbExecError`. + * call. Failing to acquire it, or losing it before any of the batch reaches the + * wire, raises `LegacyDbConnectError` (a connection failure, surfaced verbatim — + * not masked as an exec error), consistent with {@link queryRaw}; only a batch + * that was actually written raises `LegacyDbExecError`. */ readonly execBatch: ( statements: ReadonlyArray, diff --git a/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.integration.test.ts b/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.integration.test.ts index 1193d02ebf..4f3a477c10 100644 --- a/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.integration.test.ts +++ b/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.integration.test.ts @@ -176,6 +176,8 @@ const fakeBatchServer = ( readonly emptyAt?: number; /** Never answer an extended-protocol frame, so a batch hangs until interrupted. */ readonly stall?: boolean; + /** Drop the connection on the first Sync, so a batch dies mid-flight. */ + readonly destroyOnFirstSync?: boolean; } = {}, ): Promise<{ readonly port: number; @@ -268,6 +270,10 @@ const fakeBatchServer = ( } } else if (type === "S") { state.syncs += 1; + if (options.destroyOnFirstSync === true && state.syncs === 1) { + socket.destroy(); + return; + } if (options.failOnSync === true && !failed) { socket.write( errorResponse({ @@ -652,6 +658,28 @@ describe("legacyDbConnectionSqlPgLayer extended batches", () => { }), ); + it.live("fails a batch whose connection drops after it was written, then recovers", () => + // Guards the driver path that already worked: a socket dropped after the batch was + // written must surface as a batch failure, and its client must not be recycled. + Effect.gen(function* () { + const server = yield* Effect.promise(() => fakeBatchServer({ destroyOnFirstSync: true })); + yield* runWithBatchServer(server, (session) => + Effect.gen(function* () { + const error = yield* session.execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }]).pipe( + Effect.flip, + Effect.timeoutOrElse({ + duration: Duration.seconds(10), + orElse: () => Effect.die("execBatch never settled after the connection died"), + }), + ); + expect(error._tag).toBe("LegacyDbExecError"); + expect(asBatchExecError(error).message).toContain("Connection terminated unexpectedly"); + yield* session.execBatch([{ sql: "SELECT 3" }]); + }), + ); + }), + ); + it.live("classifies a failed batch-connection acquisition as a connect error", () => // A batch checks its own connection out of the pool, so a refused checkout is a // CONNECTION failure — not statement 0 failing. Misclassifying it as an exec error diff --git a/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.layer.ts b/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.layer.ts index d938298368..3f20cdd5ce 100644 --- a/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.layer.ts +++ b/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.layer.ts @@ -202,6 +202,62 @@ export function legacyToExecError(error: unknown): LegacyDbExecError { return new LegacyDbExecError({ message: String(error), code: legacyExtractSqlState(error) }); } +const LEGACY_BATCH_CONNECTION_LOST = + "connection to the database was lost before the batch could be sent"; + +/** + * pgconn's keepalive period (its default dialer, 5 minutes). Go applies that period to both + * the idle time and the probe interval; Node can only set the idle time, leaving interval and + * count at OS defaults, so a silently dead peer surfaces roughly 11 minutes later on Linux — + * sooner than Go's own window, not later. + */ +const LEGACY_PGCONN_KEEPALIVE_MILLIS = 300_000; + +/** + * Maps a failed migration batch to its public error. A batch that never reached the wire + * is a connectivity failure, not a statement failure, so it reports as one instead of + * blaming the batch's first statement; anything else keeps `legacyToExecError`'s + * server-error rendering plus the number of statements that completed. + */ +export function legacyBatchFailureError( + error: Error, + batch: + | { readonly completed: number; readonly submitted: boolean; readonly poisoned: boolean } + | undefined, +): LegacyDbExecError | LegacyDbConnectError { + if (batch === undefined || (!batch.submitted && !batch.poisoned)) { + return new LegacyDbConnectError({ + message: + error.message === LEGACY_BATCH_CONNECTION_LOST + ? LEGACY_BATCH_CONNECTION_LOST + : `${LEGACY_BATCH_CONNECTION_LOST}: ${error.message}`, + }); + } + const mapped = legacyToExecError(error); + return new LegacyDbExecError({ + message: mapped.message, + code: mapped.code, + detail: mapped.detail, + position: mapped.position, + statementIndex: batch.completed, + }); +} + +/** + * Whether a batch's pooled client must be destroyed rather than returned to the pool. A + * batch that never reached the wire leaves the client looking healthy to pg-pool while its + * socket is already gone, so the next checkout would write into the same dead connection. + */ +export function legacyShouldDiscardBatchClient( + batch: { readonly submitted: boolean } | undefined, + exit: Exit.Exit, +): boolean { + return ( + batch?.submitted === false || + (Exit.isFailure(exit) && (Cause.hasInterrupts(exit.cause) || Cause.hasDies(exit.cause))) + ); +} + const legacyEncodeTextArray = (values: ReadonlyArray): string => `{${values .map((value) => `"${value.replaceAll("\\", "\\\\").replaceAll('"', '\\"')}"`) @@ -210,7 +266,7 @@ const legacyEncodeTextArray = (values: ReadonlyArray): string => const legacyEncodeBatchValue = (value: LegacyDbBatchValue): string | null => value === null ? null : typeof value === "string" ? value : legacyEncodeTextArray(value); -class LegacyPgBatchQuery implements Pg.Submittable { +export class LegacyPgBatchQuery implements Pg.Submittable { readonly statements: ReadonlyArray<{ readonly sql: string; readonly params: ReadonlyArray; @@ -218,6 +274,7 @@ class LegacyPgBatchQuery implements Pg.Submittable { callback: (error: Error | undefined) => void; completed = 0; poisoned = false; + submitted = false; constructor( statements: ReadonlyArray, @@ -231,6 +288,9 @@ class LegacyPgBatchQuery implements Pg.Submittable { } submit(connection: Pg.Connection): Error | null { + if (!connection.stream.writable) { + return new Error(LEGACY_BATCH_CONNECTION_LOST); + } let started = false; connection.stream.cork?.(); try { @@ -242,6 +302,7 @@ class LegacyPgBatchQuery implements Pg.Submittable { connection.execute({ portal: "" }, true); } connection.sync(); + this.submitted = true; return null; } catch (error) { this.poisoned = started; @@ -511,6 +572,8 @@ export function legacyBuildRawPgConfig( : { host, port, user: cfg.user, password: cfg.password, database: cfg.database }), ...(sslOption === undefined ? {} : { ssl: sslOption }), connectionTimeoutMillis: connectTimeoutSeconds * 1000, + keepAlive: true, + keepAliveInitialDelayMillis: LEGACY_PGCONN_KEEPALIVE_MILLIS, }; } @@ -964,11 +1027,12 @@ const connect = ( // Checking a connection out of the pool for a batch is a connection-setup // concern, so it fails with `LegacyDbConnectError` — the same classification // `acquireRawClient` uses above, and for the same reason: the pool may have to - // redial (its single connection is discarded after an interrupted or poisoned - // batch), and a refused/auth/DNS failure there is not a statement failure. Mapping - // it to `LegacyDbExecError` would lose the connect suggestion and make the + // redial (its single connection is discarded after an interrupted, poisoned, or + // unsent batch), and a refused/auth/DNS failure there is not a statement failure. + // Mapping it to `LegacyDbExecError` would lose the connect suggestion and make the // migration-apply formatter blame the batch's first statement for a connectivity - // problem. Only the batch's own execution (below) raises `LegacyDbExecError`. + // problem — which is also why a batch that never reached the wire reports the same + // way (below). Only a batch that was actually written raises `LegacyDbExecError`. const acquireBatchClient = Effect.callback((resume) => { let done = false; try { @@ -1007,7 +1071,7 @@ const connect = ( (activeClient) => { const onConnectionError = () => {}; activeClient.on("error", onConnectionError); - return Effect.callback((resume) => { + return Effect.callback((resume) => { let done = false; const finish = (error: Error | undefined) => { if (done) return; @@ -1016,18 +1080,7 @@ const connect = ( resume(Effect.void); return; } - const mapped = legacyToExecError(error); - resume( - Effect.fail( - new LegacyDbExecError({ - message: mapped.message, - code: mapped.code, - detail: mapped.detail, - position: mapped.position, - statementIndex: batchQuery?.completed ?? 0, - }), - ), - ); + resume(Effect.fail(legacyBatchFailureError(error, batchQuery))); }; batchQuery = new LegacyPgBatchQuery(statements, finish); try { @@ -1046,11 +1099,8 @@ const connect = ( }, (activeClient, exit) => Effect.sync(() => { - const discard = - batchQuery?.poisoned === true || - (Exit.isFailure(exit) && - (Cause.hasInterrupts(exit.cause) || Cause.hasDies(exit.cause))); - activeClient.release(discard ? new Error("batch execution interrupted") : undefined); + const discard = legacyShouldDiscardBatchClient(batchQuery, exit); + activeClient.release(discard ? new Error("batch connection discarded") : undefined); }), ); }; diff --git a/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.unit.test.ts b/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.unit.test.ts index 4d24a158db..b42cb46a95 100644 --- a/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.unit.test.ts +++ b/apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.unit.test.ts @@ -1,14 +1,19 @@ import { EventEmitter } from "node:events"; import { Effect, Exit } from "effect"; import { SqlError, SqlSyntaxError, UnknownError } from "effect/unstable/sql/SqlError"; +import type * as Pg from "pg"; import { describe, expect, it } from "vitest"; +import { ErrorActionabilityId } from "../../shared/telemetry/error-actionability.ts"; import { legacyAcquireProbedPool, + legacyBatchFailureError, legacyBuildConnectionUrl, legacyBuildPoolConfig, legacyBuildRawPgConfig, legacyInstallPoolErrorSwallow, + LegacyPgBatchQuery, + legacyShouldDiscardBatchClient, legacyPoolStepDownVerify, legacyIsTerminalConnectError, legacyIsUnixSocketHost, @@ -322,6 +327,22 @@ describe("legacyBuildRawPgConfig", () => { const c = legacyBuildRawPgConfig({ ...base }, "h", 5432, undefined, 5); expect("ssl" in c).toBe(false); }); + + it("enables TCP keepalive with pgconn's five-minute idle delay on both config forms", () => { + const discrete = legacyBuildRawPgConfig({ ...base }, "db.example.com", 5432, false, 10); + expect(discrete.keepAlive).toBe(true); + expect(discrete.keepAliveInitialDelayMillis).toBe(300_000); + + const url = legacyBuildRawPgConfig( + { ...base, options: "reference=abc" }, + "db.example.com", + 6543, + false, + 2, + ); + expect(url.keepAlive).toBe(true); + expect(url.keepAliveInitialDelayMillis).toBe(300_000); + }); }); describe("legacyBuildPoolConfig", () => { @@ -352,6 +373,12 @@ describe("legacyBuildPoolConfig", () => { expect(c.ssl).toBe(false); }); + it("carries the raw client's TCP keepalive through to every pooled connection", () => { + const c = legacyBuildPoolConfig({ ...base }, "127.0.0.1", 54322, false, 2, false); + expect(c.keepAlive).toBe(true); + expect(c.keepAliveInitialDelayMillis).toBe(300_000); + }); + it("installs the step-down verify hook only when required", () => { // Every NEW physical connection (initial + silent redials) runs the step-down // before pg-pool hands it to a checkout, mirroring Go's per-connection @@ -561,3 +588,159 @@ describe("legacyToExecError (pg server-error extraction)", () => { expect(error.position).toBeUndefined(); }); }); + +describe("LegacyPgBatchQuery.submit", () => { + const fakeConnection = (writable: boolean) => { + const frames: Array = []; + const record = (frame: string) => () => { + frames.push(frame); + }; + return { + frames, + connection: { + stream: { writable, cork: record("cork"), uncork: record("uncork") }, + parse: record("parse"), + bind: record("bind"), + describe: record("describe"), + execute: record("execute"), + sync: record("sync"), + } as unknown as Pg.Connection, + }; + }; + + it("refuses to write a batch onto a stream that is no longer writable", () => { + const { connection, frames } = fakeConnection(false); + const batch = new LegacyPgBatchQuery([{ sql: "select 1" }], () => {}); + + const error = batch.submit(connection); + + expect(error?.message).toBe( + "connection to the database was lost before the batch could be sent", + ); + expect(batch.submitted).toBe(false); + expect(batch.poisoned).toBe(false); + expect(frames).toEqual([]); + }); + + it("writes parse/bind/describe/execute per statement and one sync while writable", () => { + const { connection, frames } = fakeConnection(true); + const batch = new LegacyPgBatchQuery([{ sql: "select 1" }, { sql: "select 2" }], () => {}); + + expect(batch.submit(connection)).toBeNull(); + + expect(frames).toEqual([ + "cork", + "parse", + "bind", + "describe", + "execute", + "parse", + "bind", + "describe", + "execute", + "sync", + "uncork", + ]); + expect(batch.submitted).toBe(true); + }); +}); + +describe("legacyBatchFailureError", () => { + it("reports an unsent batch as a lost connection rather than a statement failure", () => { + const error = legacyBatchFailureError( + new Error("connection to the database was lost before the batch could be sent"), + { completed: 0, submitted: false, poisoned: false }, + ); + + expect(error._tag).toBe("LegacyDbConnectError"); + expect(error.message).toBe( + "connection to the database was lost before the batch could be sent", + ); + expect(error[ErrorActionabilityId]).toMatchObject({ error_category: "db_connection" }); + }); + + it("keeps the driver's own reason when pg refused the batch before submit", () => { + const error = legacyBatchFailureError( + new Error("Client has encountered a connection error and is not queryable"), + { completed: 0, submitted: false, poisoned: false }, + ); + + expect(error._tag).toBe("LegacyDbConnectError"); + expect(error.message).toBe( + "connection to the database was lost before the batch could be sent: " + + "Client has encountered a connection error and is not queryable", + ); + }); + + it("reports a batch that never reached the driver the same way", () => { + const error = legacyBatchFailureError(new Error("client was closed"), undefined); + + expect(error._tag).toBe("LegacyDbConnectError"); + expect(error.message).toContain( + "connection to the database was lost before the batch could be sent", + ); + }); + + it("keeps a partially written batch on the statement path", () => { + const error = legacyBatchFailureError(new Error("serialization blew up"), { + completed: 1, + submitted: false, + poisoned: true, + }); + + expect(error._tag).toBe("LegacyDbExecError"); + }); + + it("keeps server-error mapping and the completed count for a statement failure", () => { + const error = legacyBatchFailureError( + new SqlError({ + reason: new SqlSyntaxError({ + cause: Object.assign(new Error('type "ltree" does not exist'), { + severity: "ERROR", + code: "42704", + detail: "Some detail line.", + position: "25", + }), + message: "Failed to execute statement", + operation: "execute", + }), + }), + { completed: 3, submitted: true, poisoned: false }, + ); + + expect(error._tag).toBe("LegacyDbExecError"); + expect(error).toMatchObject({ + message: 'ERROR: type "ltree" does not exist (SQLSTATE 42704)', + code: "42704", + detail: "Some detail line.", + position: 25, + statementIndex: 3, + }); + }); +}); + +describe("legacyShouldDiscardBatchClient", () => { + it("discards a client whose batch never reached the wire", () => { + expect(legacyShouldDiscardBatchClient({ submitted: false }, Exit.succeed(undefined))).toBe( + true, + ); + }); + + it("returns a client to the pool once its batch was written, error or not", () => { + expect(legacyShouldDiscardBatchClient({ submitted: true }, Exit.succeed(undefined))).toBe( + false, + ); + expect( + legacyShouldDiscardBatchClient({ submitted: true }, Exit.fail(new Error("server said no"))), + ).toBe(false); + }); + + it("discards a client whose batch was interrupted or died mid-flight", () => { + expect(legacyShouldDiscardBatchClient({ submitted: true }, Exit.interrupt(1))).toBe(true); + expect(legacyShouldDiscardBatchClient({ submitted: true }, Exit.die("boom"))).toBe(true); + }); + + it("keeps a client that was checked out but never used", () => { + expect(legacyShouldDiscardBatchClient(undefined, Exit.succeed(undefined))).toBe(false); + }); +}); diff --git a/apps/cli/src/legacy/shared/legacy-migration-apply.ts b/apps/cli/src/legacy/shared/legacy-migration-apply.ts index f19e292a1c..52eb5077d3 100644 --- a/apps/cli/src/legacy/shared/legacy-migration-apply.ts +++ b/apps/cli/src/legacy/shared/legacy-migration-apply.ts @@ -744,9 +744,9 @@ const execMigrationBatch = ( const base = executed; yield* session.execBatch(operations).pipe( Effect.mapError((cause) => { - // Acquiring the batch's connection failed: there is no failing - // statement to name, so the connect error (and its suggestion) is - // surfaced verbatim instead of being rendered as `At statement: N`. + // The batch's connection failed, either on checkout or before any of + // it reached the wire: there is no failing statement to name, so the + // connect error is surfaced verbatim instead of `At statement: N`. if (cause instanceof LegacyDbConnectError) return cause; // `statementIndex` is set by every batch failure the driver raises; a // session that omits it can only have failed before the first statement. diff --git a/apps/cli/src/legacy/shared/legacy-migration-apply.unit.test.ts b/apps/cli/src/legacy/shared/legacy-migration-apply.unit.test.ts index 9bd46cf31d..2be58dfd85 100644 --- a/apps/cli/src/legacy/shared/legacy-migration-apply.unit.test.ts +++ b/apps/cli/src/legacy/shared/legacy-migration-apply.unit.test.ts @@ -11,7 +11,7 @@ import { type CliErrorActionabilityDeclaration, ErrorActionabilityId, } from "../../shared/telemetry/error-actionability.ts"; -import type { LegacyDbConnectError } from "./legacy-db-connection.errors.ts"; +import { LegacyDbConnectError } from "./legacy-db-connection.errors.ts"; import type { LegacyDbBatchStatement, LegacyDbSession } from "./legacy-db-connection.service.ts"; import { legacyApplyMigrationFile, @@ -43,6 +43,7 @@ function fakeSession( failAfterBatch?: boolean; failWith?: { message: string; code?: string; detail?: string; position?: number }; restoreRoleSql?: string; + batchConnectionLost?: string; } = {}, ) { const calls: Array<{ @@ -65,6 +66,9 @@ function fakeSession( sql: statements.map(({ sql }) => sql).join(";\n"), statements, }); + if (opts.batchConnectionLost !== undefined) { + return Effect.fail(new LegacyDbConnectError({ message: opts.batchConnectionLost })); + } const statementIndex = opts.failAfterBatch ? statements.length : statements.findIndex(({ sql }) => @@ -217,6 +221,30 @@ describe("legacyApplyMigrationFile", () => { ); }); + it.effect("surfaces a lost batch connection verbatim, never as a failing statement", () => { + const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); + const file = join(dir, "20240101120000_lost.sql"); + writeFileSync(file, "ALTER TABLE a ADD COLUMN b int;"); + const { session } = fakeSession({ + batchConnectionLost: "connection to the database was lost before the batch could be sent", + }); + return run(session, file).pipe( + Effect.flip, + Effect.tap((error) => + Effect.sync(() => { + expect(error).toBeInstanceOf(LegacyDbConnectError); + expect(error.message).toBe( + "connection to the database was lost before the batch could be sent", + ); + const rendered = JSON.stringify(error); + expect(rendered).not.toContain("At statement:"); + expect(rendered).not.toContain("ALTER TABLE a ADD COLUMN b int"); + rmSync(dir, { recursive: true, force: true }); + }), + ), + ); + }); + it.effect("sends a large compatible migration in one batch", () => { const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); const file = join(dir, "20240101120000_many.sql");