fix(cli): detect dead db connections (CLI-2207)

apps/cli/docs/go-cli-divergences.md

@@ -246,3 +246,13 @@ These commands exist in the TS CLI today but have no direct top-level equivalent
   an output channel, and in managed mode inside the daemon process, so a user-visible warning
   there needs a diagnostics channel on `BuildResult` first; the applied value stays visible via
   `docker inspect`.
+- 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 `<err>\nAt statement: N\n<sql>`. 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. This covers batches only: a
+  batch that was written, and the pipeline-incompatible statements the same loop runs
+  standalone through `exec` (`CREATE INDEX CONCURRENTLY`, `VACUUM`, …), both keep Go's
+  `At statement: N` rendering for every failure, transport included.

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<LegacyDbBatchStatement>,

apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.integration.test.ts

@@ -20,6 +20,7 @@ import {
 import {
   legacyAcquirePgPool,
   legacyDbConnectionSqlPgLayer,
+  LegacyPgBatchQuery,
 } from "./legacy-db-connection.sql-pg.layer.ts";
 
 const SUGGESTION_CONTEXT = {
@@ -176,11 +177,14 @@ 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;
   readonly close: () => void;
   readonly state: FakeBatchServerState;
+  readonly sockets: ReadonlyArray<net.Socket>;
 }> =>
   new Promise((resolve) => {
     const state: FakeBatchServerState = {
@@ -189,7 +193,9 @@ const fakeBatchServer = (
       params: [],
       syncs: 0,
     };
+    const sockets: Array<net.Socket> = [];
     const server = net.createServer((socket) => {
+      sockets.push(socket);
       let sawStartup = false;
       let pending = Buffer.alloc(0);
       let failed = false;
@@ -268,6 +274,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({
@@ -288,7 +298,7 @@ const fakeBatchServer = (
     });
     server.listen(0, "127.0.0.1", () => {
       const address = server.address() as net.AddressInfo;
-      resolve({ port: address.port, close: () => server.close(), state });
+      resolve({ port: address.port, close: () => server.close(), state, sockets });
     });
   });
 
@@ -652,6 +662,100 @@ describe("legacyDbConnectionSqlPgLayer extended batches", () => {
     }),
   );
 
+  it.live("fails a batch whose connection drops after it was written, then recovers", () =>
+    // A socket dropped after the batch was written must fail that batch and must not leave
+    // the client to be handed to the next one.
+    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("survives an idle raw-client socket death and redials for the next query", () =>
+    // node-postgres emits `error` on an idle client; with no listener that terminates the
+    // process, so a database that dies between two `queryRaw` calls must fail that call
+    // rather than the CLI, and must not leave the corpse cached for the call after it.
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() => fakeBatchServer());
+      yield* runWithBatchServer(server, (session) =>
+        Effect.gen(function* () {
+          yield* session.queryRaw("SELECT 1");
+          // `queryRaw` runs on its own client, opened after the pool's, so it is the
+          // newest connection the server has accepted.
+          const rawSocket = server.sockets.at(-1);
+          const openedBeforeRedial = server.sockets.length;
+
+          const closed = new Promise<void>((resolve) => {
+            rawSocket?.on("close", () => resolve());
+          });
+          yield* Effect.sync(() => rawSocket?.destroy());
+          yield* Effect.promise(() => closed);
+
+          const failed = yield* session.queryRaw("SELECT 1").pipe(
+            Effect.flip,
+            Effect.timeoutOrElse({
+              duration: Duration.seconds(10),
+              orElse: () => Effect.die("queryRaw never settled after the idle socket died"),
+            }),
+          );
+          expect(failed._tag).toBe("LegacyDbExecError");
+
+          yield* session.queryRaw("SELECT 1");
+          expect(server.sockets.length).toBeGreaterThan(openedBeforeRedial);
+        }),
+      );
+    }),
+  );
+
+  it.live("refuses to write a batch onto a real pooled client whose socket is already gone", () =>
+    // The unit test drives `submit` through a hand-built connection; this pins the same
+    // refusal against a real node-postgres client, so a driver change that stops making the
+    // socket unwritable would be caught rather than mocked over. Destroying the socket and
+    // submitting in one synchronous block keeps the window deterministic: `writable` flips
+    // immediately, while pg only marks the client unqueryable on the next tick's close.
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() => fakeBatchServer());
+      yield* Effect.gen(function* () {
+        const pool = yield* legacyAcquirePgPool(
+          {
+            host: "127.0.0.1",
+            port: server.port,
+            user: "postgres",
+            password: SENTINEL_PASSWORD,
+            database: "postgres",
+            sslmode: "disable",
+          },
+          { isLocal: true, dnsResolver: "native" },
+        );
+        const client = yield* Effect.promise(() => pool.connect());
+        const batch = new LegacyPgBatchQuery([{ sql: "SELECT 1" }], () => {});
+
+        const refusal = yield* Effect.sync(() => {
+          client.connection.stream.destroy();
+          return batch.submit(client.connection);
+        });
+
+        expect(refusal?.message).toBe("the connection's socket is no longer writable");
+        expect(batch.outcome).toBe("unsent");
+        expect(server.state.frameTypes).toEqual([]);
+        client.release(new Error("done"));
+      }).pipe(Effect.scoped, Effect.ensuring(Effect.sync(server.close)));
+    }),
+  );
+
   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

apps/cli/src/legacy/shared/legacy-db-connection.sql-pg.layer.ts

@@ -12,6 +12,7 @@ import { ConnectionError, SqlError } from "effect/unstable/sql/SqlError";
 import * as Pg from "pg";
 import { to as pgCopyTo } from "pg-copy-streams";
 import {
+  LEGACY_SUGGEST_LOCAL_STACK,
   legacyConnectFailureMessage,
   legacyConnectSuggestion,
   legacyIsDialFailure,
@@ -202,6 +203,69 @@ 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";
+
+/** How far a batch got on the wire: nothing sent, a partial write, or fully written. */
+export type LegacyBatchOutcome = "unsent" | "poisoned" | "submitted";
+
+/**
+ * pgconn's keepalive period (its default dialer, 5 minutes). Go applies that period to both
+ * the idle time and the probe interval; Node only sets the idle time and leaves the probe
+ * schedule to the runtime and OS, so a silently dead peer surfaces sooner here than under Go.
+ */
+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 outcome: LegacyBatchOutcome } | undefined,
+  isLocal: boolean,
+): LegacyDbExecError | LegacyDbConnectError {
+  if (batch === undefined || batch.outcome === "unsent") {
+    return new LegacyDbConnectError({
+      message: `${LEGACY_BATCH_CONNECTION_LOST}: ${error.message}`,
+      // The checkout failure a tick earlier carries this same hint, so losing the
+      // connection mid-batch must not silently drop it.
+      ...(isLocal ? { suggestion: LEGACY_SUGGEST_LOCAL_STACK } : {}),
+    });
+  }
+  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.
+ *
+ * A batch that WAS written keeps its client: a statement failure should not cost a redial and
+ * a fresh step-down on a single-connection pool. Recovering from a socket that died after the
+ * write is left to pg-pool, which drops a released client whose private `_queryable` flag is
+ * false — so that is the behavior to re-check if a pg-pool bump ever breaks the recovery this
+ * layer's integration tests assert.
+ */
+export function legacyShouldDiscardBatchClient(
+  batch: { readonly outcome: LegacyBatchOutcome } | undefined,
+  exit: Exit.Exit<unknown, unknown>,
+): boolean {
+  return (
+    (batch !== undefined && batch.outcome !== "submitted") ||
+    (Exit.isFailure(exit) && (Cause.hasInterrupts(exit.cause) || Cause.hasDies(exit.cause)))
+  );
+}
+
 const legacyEncodeTextArray = (values: ReadonlyArray<string>): string =>
   `{${values
     .map((value) => `"${value.replaceAll("\\", "\\\\").replaceAll('"', '\\"')}"`)
@@ -210,14 +274,14 @@ const legacyEncodeTextArray = (values: ReadonlyArray<string>): 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<string | null>;
   }>;
   callback: (error: Error | undefined) => void;
   completed = 0;
-  poisoned = false;
+  outcome: LegacyBatchOutcome = "unsent";
 
   constructor(
     statements: ReadonlyArray<LegacyDbBatchStatement>,
@@ -231,6 +295,9 @@ class LegacyPgBatchQuery implements Pg.Submittable {
   }
 
   submit(connection: Pg.Connection): Error | null {
+    if (!connection.stream.writable) {
+      return new Error("the connection's socket is no longer writable");
+    }
     let started = false;
     connection.stream.cork?.();
     try {
@@ -242,9 +309,10 @@ class LegacyPgBatchQuery implements Pg.Submittable {
         connection.execute({ portal: "" }, true);
       }
       connection.sync();
+      this.outcome = "submitted";
       return null;
     } catch (error) {
-      this.poisoned = started;
+      this.outcome = started ? "poisoned" : "unsent";
       return error instanceof Error ? error : new Error(String(error));
     } finally {
       connection.stream.uncork?.();
@@ -511,6 +579,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,
   };
 }
 
@@ -946,6 +1016,12 @@ const connect = (
     const acquireRawClient = Effect.gen(function* () {
       if (rawClient !== undefined) return rawClient;
       const fresh = new Pg.Client(winningRawConfig);
+      // node-postgres emits `error` on a cached client whose socket dies while idle; with
+      // no listener that terminates the process, so absorb it and drop the dead client so
+      // the next acquisition redials instead of reusing it.
+      fresh.on("error", () => {
+        if (rawClient === fresh) rawClient = undefined;
+      });
       yield* Effect.tryPromise({
         try: () => fresh.connect(),
         catch: (error) => legacyToConnectError(cfg, options.isLocal, error),
@@ -964,11 +1040,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<Pg.PoolClient, LegacyDbConnectError>((resume) => {
       let done = false;
       try {
@@ -1007,7 +1084,7 @@ const connect = (
         (activeClient) => {
           const onConnectionError = () => {};
           activeClient.on("error", onConnectionError);
-          return Effect.callback<void, LegacyDbExecError>((resume) => {
+          return Effect.callback<void, LegacyDbExecError | LegacyDbConnectError>((resume) => {
             let done = false;
             const finish = (error: Error | undefined) => {
               if (done) return;
@@ -1016,18 +1093,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, options.isLocal)));
             };
             batchQuery = new LegacyPgBatchQuery(statements, finish);
             try {
@@ -1046,11 +1112,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);
           }),
       );
     };