fix(cli): make migration batches transactional (CLI-2261)

apps/cli/src/command-internal/db-connection.errors.ts

@@ -36,6 +36,12 @@ export class DbExecError extends Data.TaggedError("DbExecError")<{
    * the batch length for a deferred Sync failure. Absent for `exec`/`query`.
    */
   readonly statementIndex?: number;
+  /**
+   * Which CLI-injected transaction wrapper was in flight when the batch failed —
+   * whether the server rejected BEGIN/COMMIT or the connection was lost while one
+   * was pending — rather than a caller statement. Absent otherwise.
+   */
+  readonly transactionPhase?: "begin" | "commit";
   /**
    * Postgres SQLSTATE (e.g. `42P01` undefined_table), extracted from the driver
    * error's `cause` chain when present. Lets callers match Go's error-code checks

apps/cli/src/command-internal/db-connection.service.ts

@@ -104,9 +104,12 @@ export interface DbSession {
   /** Run a single SQL statement, ignoring any returned rows. */
   readonly exec: (sql: string) => Effect.Effect<void, DbExecError>;
   /**
-   * Run statements as one extended-protocol batch with a single final Sync.
-   * On failure, {@link DbExecError.statementIndex} is the number of
-   * statements that completed before the error.
+   * Run statements as one extended-protocol batch inside a single explicit
+   * transaction, with a single final Sync — a bare pipeline is not a transaction
+   * block (supabase/cli#6347). On failure a bounded, best-effort rollback runs
+   * before the connection can be reused (a rollback that does not succeed
+   * discards the connection), and {@link DbExecError.statementIndex} is
+   * the number of the caller's 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, or losing it before any of the batch reaches the

apps/cli/src/command-internal/db-connection.sql-pg.integration.test.ts

@@ -11,7 +11,7 @@ import { describe, expect, it } from "@effect/vitest";
 import { Duration, Effect } from "effect";
 
 import { SUGGEST_ENV_VAR, SUGGEST_LOCAL_STACK } from "./connect-errors.ts";
-import type { DbConnectError, DbExecError } from "./db-connection.errors.ts";
+import { type DbConnectError, DbExecError } from "./db-connection.errors.ts";
 import { type DbSession, type PgConnInput, DbConnection } from "./db-connection.service.ts";
 import {
   acquirePgPool,
@@ -162,6 +162,7 @@ interface FakeBatchServerState {
   readonly frameTypes: Array<string>;
   readonly statements: Array<string>;
   readonly params: Array<ReadonlyArray<string | null>>;
+  readonly simpleQueries: Array<string>;
   syncs: number;
 }
 
@@ -173,8 +174,12 @@ const fakeBatchServer = (
     readonly emptyAt?: number;
     /** Never answer an extended-protocol frame, so a batch hangs until interrupted. */
     readonly stall?: boolean;
+    readonly stallRollback?: boolean;
+    readonly destroyOnRollback?: boolean;
     /** Drop the connection on the first Sync, so a batch dies mid-flight. */
     readonly destroyOnFirstSync?: boolean;
+    /** Drop the connection at the first Execute (BEGIN's), before anything completes. */
+    readonly destroyOnFirstExecute?: boolean;
   } = {},
 ): Promise<{
   readonly port: number;
@@ -187,9 +192,11 @@ const fakeBatchServer = (
       frameTypes: [],
       statements: [],
       params: [],
+      simpleQueries: [],
       syncs: 0,
     };
     const sockets: Array<net.Socket> = [];
+    let destroyedOnExecute = false;
     const server = net.createServer((socket) => {
       sockets.push(socket);
       let sawStartup = false;
@@ -219,6 +226,13 @@ const fakeBatchServer = (
           const body = pending.subarray(5, length + 1);
           pending = pending.subarray(length + 1);
           if (type === "Q") {
+            const sql = body.toString("utf8", 0, body.length - 1);
+            state.simpleQueries.push(sql);
+            if (options.stallRollback === true && sql === "ROLLBACK") continue;
+            if (options.destroyOnRollback === true && sql === "ROLLBACK") {
+              socket.destroy();
+              return;
+            }
             socket.write(Buffer.concat([commandComplete("SELECT 1"), READY_FOR_QUERY]));
             continue;
           }
@@ -250,6 +264,11 @@ const fakeBatchServer = (
           } else if (type === "D") {
             if (!failed) socket.write(NO_DATA);
           } else if (type === "E") {
+            if (options.destroyOnFirstExecute === true && !destroyedOnExecute) {
+              destroyedOnExecute = true;
+              socket.destroy();
+              return;
+            }
             if (!failed) {
               if (activeIndex === options.failExecuteAt) {
                 failed = true;
@@ -547,9 +566,9 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
       Effect.ensuring(Effect.sync(server.close)),
     );
 
-  it.live("sends every statement and parameter set before one Sync", () =>
+  it.live("sends every statement and parameter set inside one BEGIN/COMMIT before one Sync", () =>
     Effect.gen(function* () {
-      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 1 }));
+      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 2 }));
       const values = ["plain", 'quote"', "slash\\", "comma,", "{brace}", "line\nbreak", "NULL", ""];
       yield* runWithBatchServer(server, (session) =>
         session.execBatch([
@@ -562,18 +581,22 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
         ]),
       );
       expect(server.state.statements).toEqual([
+        "BEGIN",
         "SELECT 1",
         "-- comment only",
         "INSERT INTO history(version, name, statements) VALUES($1, $2, $3)",
+        "COMMIT",
       ]);
       expect(server.state.params).toEqual([
+        [],
         [],
         [],
         [
           "v'1",
           "name\\two",
           '{"plain","quote\\\"","slash\\\\","comma,","{brace}","line\nbreak","NULL",""}',
         ],
+        [],
       ]);
       expect(server.state.frameTypes).toEqual([
         "P",
@@ -588,15 +611,24 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
         "B",
         "D",
         "E",
+        "P",
+        "B",
+        "D",
+        "E",
+        "P",
+        "B",
+        "D",
+        "E",
         "S",
       ]);
       expect(server.state.syncs).toBe(1);
+      expect(server.state.simpleQueries).not.toContain("ROLLBACK");
     }),
   );
 
   it.live("maps a later parse failure to its statement and keeps its local position", () =>
     Effect.gen(function* () {
-      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 1, failAt: 2 }));
+      const server = yield* Effect.promise(() => fakeBatchServer({ emptyAt: 2, failAt: 3 }));
       yield* runWithBatchServer(server, (session) =>
         Effect.gen(function* () {
           const error = asBatchExecError(
@@ -617,7 +649,7 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
 
   it.live("maps a position-less runtime failure from completed commands", () =>
     Effect.gen(function* () {
-      const server = yield* Effect.promise(() => fakeBatchServer({ failExecuteAt: 1 }));
+      const server = yield* Effect.promise(() => fakeBatchServer({ failExecuteAt: 2 }));
       yield* runWithBatchServer(server, (session) =>
         session
           .execBatch([{ sql: "SELECT 1" }, { sql: "INSERT duplicate" }, { sql: "SELECT 3" }])
@@ -655,6 +687,72 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
     }),
   );
 
+  it.live("rolls a failed batch's transaction back before the pooled client is reused", () =>
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() => fakeBatchServer({ failExecuteAt: 3 }));
+      yield* runWithBatchServer(server, (session) =>
+        Effect.gen(function* () {
+          yield* session
+            .execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }, { sql: "INSERT duplicate" }])
+            .pipe(Effect.flip);
+          const sockets = server.sockets.length;
+          yield* session.execBatch([{ sql: "SELECT 3" }]);
+          expect(server.sockets.length).toBe(sockets);
+        }),
+      );
+      expect(server.state.simpleQueries).toContain("ROLLBACK");
+      expect(server.state.syncs).toBe(2);
+    }),
+  );
+
+  it.live("absorbs a socket death during the error-path rollback instead of crashing", () =>
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() =>
+        fakeBatchServer({ failExecuteAt: 3, destroyOnRollback: true }),
+      );
+      yield* runWithBatchServer(server, (session) =>
+        Effect.gen(function* () {
+          yield* session
+            .execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }, { sql: "INSERT duplicate" }])
+            .pipe(Effect.flip);
+          yield* session.execBatch([{ sql: "SELECT 3" }]).pipe(
+            Effect.timeoutOrElse({
+              duration: Duration.seconds(10),
+              orElse: () => Effect.die("the batch after a dead-rollback socket never settled"),
+            }),
+          );
+          expect(server.state.simpleQueries).toContain("ROLLBACK");
+        }),
+      );
+    }),
+  );
+
+  it.live(
+    "bounds a stalled failed-batch rollback and discards the client instead of reusing it",
+    () =>
+      Effect.gen(function* () {
+        const server = yield* Effect.promise(() =>
+          fakeBatchServer({ failExecuteAt: 3, stallRollback: true }),
+        );
+        yield* runWithBatchServer(server, (session) =>
+          Effect.gen(function* () {
+            const before = server.sockets.length;
+            yield* session
+              .execBatch([{ sql: "SELECT 1" }, { sql: "SELECT 2" }, { sql: "INSERT duplicate" }])
+              .pipe(Effect.flip);
+            yield* session.execBatch([{ sql: "SELECT 3" }]).pipe(
+              Effect.timeoutOrElse({
+                duration: Duration.seconds(10),
+                orElse: () => Effect.die("the batch after a stalled rollback never settled"),
+              }),
+            );
+            expect(server.state.simpleQueries).toContain("ROLLBACK");
+            expect(server.sockets.length).toBeGreaterThan(before);
+          }),
+        );
+      }),
+  );
+
   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.
@@ -671,6 +769,35 @@ describe("dbConnectionSqlPgLayer extended batches", () => {
           );
           expect(error._tag).toBe("DbExecError");
           expect(asBatchExecError(error).message).toContain("Connection terminated unexpectedly");
+          // This server acks every statement and dies at Sync, so the loss lands
+          // on COMMIT — marked so the formatter never blames a caller statement.
+          expect(asBatchExecError(error).transactionPhase).toBe("commit");
+          yield* session.execBatch([{ sql: "SELECT 3" }]);
+        }),
+      );
+    }),
+  );
+
+  it.live("marks the begin phase when the connection drops before BEGIN completes", () =>
+    // The loss arrives while BEGIN is still in flight, so no caller statement ran:
+    // the phase marker keeps formatters from rendering `At statement: 0` for it.
+    Effect.gen(function* () {
+      const server = yield* Effect.promise(() => fakeBatchServer({ destroyOnFirstExecute: 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).toBeInstanceOf(DbExecError);
+          expect(asBatchExecError(error).message).toContain("Connection terminated unexpectedly");
+          expect(asBatchExecError(error)).toMatchObject({
+            statementIndex: 0,
+            transactionPhase: "begin",
+          });
           yield* session.execBatch([{ sql: "SELECT 3" }]);
         }),
       );