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

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 `<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. A batch that was written
+  keeps Go's rendering unchanged.

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

@@ -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

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<unknown, unknown>,
+): boolean {
+  return (
+    batch?.submitted === false ||
+    (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 +266,15 @@ 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;
+  submitted = false;
 
   constructor(
     statements: ReadonlyArray<LegacyDbBatchStatement>,
@@ -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<Pg.PoolClient, LegacyDbConnectError>((resume) => {
       let done = false;
       try {
@@ -1007,7 +1071,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 +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);
           }),
       );
     };