fix(cli): run pipeline-incompatible migration statements outside the transaction (#5139)

Migrations containing CREATE INDEX CONCURRENTLY (and VACUUM, REINDEX
CONCURRENTLY, ALTER SYSTEM, CLUSTER) failed with SQLSTATE 25001 because the TS
apply wraps every statement in a single BEGIN/COMMIT, and those statements
cannot run inside a transaction block.
Port the fix from the Go PR #5156 into the native TS apply: detect
pipeline-incompatible statements (legacyIsPipelineIncompatible), flush the open
batch, run the statement standalone, then resume batching. The history insert
goes in the final batch, so the migration is recorded only after every
statement succeeds. Migrations without such statements stay a single
BEGIN/COMMIT — behaviour is unchanged for them.
Shared by migration up/down and declarative sync.

Author: Coly010

apps/cli/src/legacy/shared/legacy-migration-apply.ts

@@ -17,13 +17,82 @@ export class LegacyMigrationApplyError extends Data.TaggedError("LegacyMigration
 readonlymessage: string;
 }>{}
 
+// Byte order mark (U+FEFF) — stripped from the head of a statement like Go does.
+constBOM_CODE_POINT=0xfeff;
+
+// Statements that PostgreSQL refuses to run inside a transaction block / extended-query
+// pipeline (SQLSTATE 25001). Ports of Go's pattern set in `pkg/migration/file.go`
+// (supabase/cli#5156). Matched against the upper-cased, comment-stripped statement.
+constCREATE_INDEX_CONCURRENTLY_PATTERN=/^CREATE\s+(?:UNIQUE\s+)?INDEX\s+CONCURRENTLY(?:\s|$)/u;
+constREINDEX_CONCURRENTLY_PATTERN=/^REINDEX(?:\s|\().*\sCONCURRENTLY(?:\s|$)/u;
+constVACUUM_PATTERN=/^VACUUM(?:\s|\(|$)/u;
+constALTER_SYSTEM_PATTERN=/^ALTER\s+SYSTEM(?:\s|$)/u;
+constCLUSTER_PATTERN=/^CLUSTER(?:\s|$)/u;
+
+/**
+ * Strips a leading BOM, whitespace, and SQL line (`--`) and block comments from the
+ * front of a statement so the keyword check below sees the real first token.
+ * Port of Go's `trimLeadingSQLComments` (`pkg/migration/file.go`, supabase/cli#5156).
+ */
+constlegacyTrimLeadingSqlComments=(sql: string): string=>{
+// Go's `TrimLeftFunc` drops a leading BOM together with whitespace; strip the BOM
+// via its code point so no irregular whitespace lands in the source.
+lettrimmed=sql.replace(/^[\t\n\r]+/u,"");
+while(trimmed.charCodeAt(0)===BOM_CODE_POINT){
+trimmed=trimmed.slice(1).replace(/^[\t\n\r]+/u,"");
+}
+for(;;){
+if(trimmed.startsWith("--")){
+constidx=trimmed.indexOf("\n");
+if(idx<0)return"";
+trimmed=trimmed.slice(idx+1).replace(/^[\t\n\r]+/u,"");
+}elseif(trimmed.startsWith("/*")){
+constidx=trimmed.indexOf("*/");
+if(idx<0)returntrimmed;
+trimmed=trimmed.slice(idx+2).replace(/^[\t\n\r]+/u,"");
+}else{
+returntrimmed.trim();
+}
+}
+};
+
+/**
+ * Whether a migration statement cannot run inside a transaction block — `CREATE
+ * [UNIQUE] INDEX CONCURRENTLY`, `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`,
+ * `CLUSTER`. Such statements fail with SQLSTATE 25001 inside the `BEGIN`/`COMMIT`
+ * that wraps a migration, so `legacyApplyMigrationFile` runs them standalone.
+ * Port of Go's `isPipelineIncompatible` (`pkg/migration/file.go`, supabase/cli#5156).
+ */
+exportconstlegacyIsPipelineIncompatible=(sql: string): boolean=>{
+constupper=legacyTrimLeadingSqlComments(sql).toUpperCase();
+return(
+CREATE_INDEX_CONCURRENTLY_PATTERN.test(upper)||
+REINDEX_CONCURRENTLY_PATTERN.test(upper)||
+VACUUM_PATTERN.test(upper)||
+ALTER_SYSTEM_PATTERN.test(upper)||
+CLUSTER_PATTERN.test(upper)
+);
+};
+
+/** A buffered statement awaiting the next batch flush; `version` is the history insert. */
+typeLegacyBatchItem=
+|{readonlykind: "exec";readonlysql: string}
+|{readonlykind: "version"};
+
 /**
  * Applies a single migration file to the connected database and records it in
  * `supabase_migrations.schema_migrations`. Mirrors Go's `migration.ApplyMigrations`
  * for one file (`pkg/migration/apply.go` + `(*MigrationFile).ExecBatch`): create
  * the history table, `RESET ALL`, then run the file's statements + the history
- * insert atomically. The whole file is one transaction (Go's `ExecBatch` is
- * implicitly transactional); on failure the transaction is rolled back.
+ * insert.
+ *
+ * Statements run inside a `BEGIN`/`COMMIT` batch, except pipeline-incompatible ones
+ * (`legacyIsPipelineIncompatible` — `CREATE INDEX CONCURRENTLY`, `VACUUM`, …) which
+ * cannot run in a transaction block: the batch is flushed (committed), the statement
+ * runs standalone, then batching resumes — mirroring Go's `ExecBatch` flush logic
+ * (supabase/cli#5156). The history insert goes in the final batch, so the migration
+ * is recorded only after every statement succeeds. A file with no such statements is
+ * a single `BEGIN`/`COMMIT` around everything, identical to the pre-fix behaviour.
  *
  * `mapError` lets the caller tag the failure (e.g. `LegacyDeclarativeApplyError`).
  */
@@ -44,8 +113,8 @@ export const legacyApplyMigrationFile = <E>(
 
 yield*legacyCreateMigrationTable(session);
 yield*session.exec("RESET ALL");
-yield*session.exec("BEGIN");
-// Mirror Go's `MigrationFile.ExecBatch` error context (`pkg/migration/file.go:88-113`):
+
+// Mirror Go's `MigrationFile.ExecBatch` error context (`pkg/migration/file.go`):
 // on a failed statement, append `At statement: <index>` and the statement text so the
 // error (and the debug bundle) point at the exact failing SQL. (Go also adds a caret /
 // pgErr.Detail / extension-type hint, which need the driver SQLSTATE the session does
@@ -56,26 +125,59 @@ export const legacyApplyMigrationFile = <E>(
  : String(e);
 constatStatement=(e: unknown,index: number,stat: string)=>
 newError(`${errMessage(e)}\nAt statement: ${index}\n${stat}`);
-constbody=Effect.gen(function*(){
-for(leti=0;i<statements.length;i++){
-conststatement=statements[i]??"";
+
+// `executed` is the global statement index of the next statement to run, so the
+// error context stays accurate across flushed batches and standalone statements
+// (Go threads the same counter through `ExecBatch`).
+letpending: ReadonlyArray<LegacyBatchItem>=[];
+letexecuted=0;
+
+constflushBatch=Effect.gen(function*(){
+if(pending.length===0)return;
+constitems=pending;
+pending=[];
+constbase=executed;
+constbody=Effect.gen(function*(){
+for(const[offset,item]ofitems.entries()){
+constindex=base+offset;
+if(item.kind==="version"){
+// Go defaults to the version-insert statement when all listed statements succeed.
+yield*session
+.query(INSERT_MIGRATION_VERSION,[version,name,statements])
+.pipe(
+Effect.mapError((cause)=>atStatement(cause,index,INSERT_MIGRATION_VERSION)),
+);
+}else{
+yield*session
+.exec(item.sql)
+.pipe(Effect.mapError((cause)=>atStatement(cause,index,item.sql)));
+}
+}
+yield*session.exec("COMMIT");
+});
+yield*session.exec("BEGIN");
+yield*body.pipe(Effect.tapError(()=>session.exec("ROLLBACK").pipe(Effect.ignore)));
+executed+=items.length;
+});
+
+for(conststatementofstatements){
+if(legacyIsPipelineIncompatible(statement)){
+// Flush the open batch, then run the incompatible statement on its own (no
+// surrounding transaction) so PostgreSQL accepts it.
+yield*flushBatch;
+constindex=executed;
 yield*session
 .exec(statement)
-.pipe(Effect.mapError((cause)=>atStatement(cause,i,statement)));
-}
-if(version.length>0){
-// Go defaults to the version-insert statement when all listed statements succeed.
-yield*session
-.query(INSERT_MIGRATION_VERSION,[version,name,statements])
-.pipe(
-Effect.mapError((cause)=>
-atStatement(cause,statements.length,INSERT_MIGRATION_VERSION),
-),
-);
+.pipe(Effect.mapError((cause)=>atStatement(cause,index,statement)));
+executed+=1;
+}else{
+pending=[...pending,{kind: "exec",sql: statement}];
 }
-yield*session.exec("COMMIT");
-});
-yield*body.pipe(Effect.tapError(()=>session.exec("ROLLBACK").pipe(Effect.ignore)));
+}
+if(version.length>0){
+pending=[...pending,{kind: "version"}];
+}
+yield*flushBatch;
 }).pipe(
 Effect.mapError((error)=>
 mapError(

apps/cli/src/legacy/shared/legacy-migration-apply.unit.test.ts

@@ -6,7 +6,10 @@ import { describe, expect, it } from "@effect/vitest";
 import{Data,Effect,Exit,FileSystem,Path}from"effect";
 
 importtype{LegacyDbSession}from"./legacy-db-connection.service.ts";
-import{legacyApplyMigrationFile}from"./legacy-migration-apply.ts";
+import{
+legacyApplyMigrationFile,
+legacyIsPipelineIncompatible,
+}from"./legacy-migration-apply.ts";
 
 classTestErrorextendsData.TaggedError("TestError")<{readonlymessage: string}>{}
 
@@ -103,4 +106,116 @@ describe("legacyApplyMigrationFile", () => {
 ),
 );
 });
+
+it.effect("runs a pipeline-incompatible statement outside the surrounding transaction",()=>{
+constdir=mkdtempSync(join(tmpdir(),"legacy-apply-"));
+constfile=join(dir,"20240101120000_add_index.sql");
+writeFileSync(
+file,
+"create table a (id int);\nCREATE INDEX CONCURRENTLY a_idx ON a(id);\nALTER TABLE a ENABLE ROW LEVEL SECURITY;",
+);
+const{ session, calls }=fakeSession();
+returnrun(session,file).pipe(
+Effect.tap(()=>
+Effect.sync(()=>{
+constexecs=calls.filter((c)=>c.kind==="exec").map((c)=>c.sql);
+constconcurrently="CREATE INDEX CONCURRENTLY a_idx ON a(id)";
+expect(execs).toContain(concurrently);
+// The CONCURRENTLY statement must not run inside an open transaction, or
+// PostgreSQL rejects it (SQLSTATE 25001). The batch is flushed first, so the
+// BEGIN/COMMIT counts before it must balance (no open transaction).
+constbefore=execs.slice(0,execs.indexOf(concurrently));
+expect(before.filter((s)=>s==="BEGIN").length).toBe(
+before.filter((s)=>s==="COMMIT").length,
+);
+// The compatible statements still ran inside a transaction...
+expect(before).toContain("BEGIN");
+expect(before).toContain("COMMIT");
+// ...and the trailing compatible statement reopens a transaction after it.
+constafter=execs.slice(execs.indexOf(concurrently)+1);
+expect(after).toContain("BEGIN");
+expect(after.indexOf("ALTER TABLE a ENABLE ROW LEVEL SECURITY")).toBeGreaterThanOrEqual(
+0,
+);
+// The migration is still recorded once every statement succeeds.
+constinsert=calls.find((c)=>c.kind==="query");
+expect(insert?.params?.[0]).toBe("20240101120000");
+rmSync(dir,{recursive: true,force: true});
+}),
+),
+);
+});
+
+it.effect("reports a pipeline-incompatible statement failure with its statement index",()=>{
+constdir=mkdtempSync(join(tmpdir(),"legacy-apply-"));
+constfile=join(dir,"20240101120000_add_index.sql");
+writeFileSync(file,"create table a (id int);\nCREATE INDEX CONCURRENTLY a_idx ON a(id);");
+const{ session, calls }=fakeSession({failOn: "CONCURRENTLY"});
+returnrun(session,file).pipe(
+Effect.exit,
+Effect.tap((exit)=>
+Effect.sync(()=>{
+expect(Exit.isFailure(exit)).toBe(true);
+if(Exit.isFailure(exit)){
+constmsg=JSON.stringify(exit.cause);
+// Index 1: the leading `create table a` (index 0) committed in its own batch first.
+expect(msg).toContain("At statement: 1");
+expect(msg).toContain("CREATE INDEX CONCURRENTLY a_idx ON a(id)");
+}
+// The migration version is not recorded when a statement fails.
+expect(calls.some((c)=>c.kind==="query")).toBe(false);
+rmSync(dir,{recursive: true,force: true});
+}),
+),
+);
+});
+});
+
+describe("legacyIsPipelineIncompatible",()=>{
+// Mirrors Go's `TestIsPipelineIncompatible` (`pkg/migration/file_test.go`, supabase/cli#5156).
+constcases: ReadonlyArray<readonly[string,string,boolean]>=[
+[
+"create index concurrently",
+"CREATE INDEX CONCURRENTLY widgets_id_idx ON public.widgets(id)",
+true,
+],
+[
+"create unique index concurrently",
+"CREATE UNIQUE INDEX CONCURRENTLY widgets_id_idx ON public.widgets(id)",
+true,
+],
+[
+"create index concurrently after comments",
+"-- cannot run in a transaction\n/* generated */\nCREATE INDEX CONCURRENTLY widgets_id_idx ON public.widgets(id)",
+true,
+],
+["reindex table concurrently","REINDEX TABLE CONCURRENTLY public.widgets",true],
+[
+"reindex with options concurrently",
+"REINDEX (VERBOSE) INDEX CONCURRENTLY widgets_id_idx",
+true,
+],
+["vacuum bare","VACUUM",true],
+["vacuum with options","VACUUM (FULL, ANALYZE) public.widgets",true],
+["alter system","ALTER SYSTEM SET wal_level = 'logical'",true],
+["cluster","CLUSTER public.widgets USING widgets_id_idx",true],
+[
+"lower-case create index concurrently",
+"create index concurrently widgets_id_idx on public.widgets(id)",
+true,
+],
+["leading whitespace before concurrently"," CREATE INDEX CONCURRENTLY a_idx ON a(id)",true],
+// Negatives — compatible statements that must keep running inside the batch transaction.
+["plain create index","CREATE INDEX widgets_id_idx ON public.widgets(id)",false],
+["create table","create table public.widgets(id bigint primary key)",false],
+["reindex without concurrently","REINDEX TABLE public.widgets",false],
+["vacuum-prefixed identifier","VACUUMING analytics",false],
+["concurrently as a column name","CREATE TABLE t (concurrently int)",false],
+["insert","INSERT INTO public.widgets VALUES (1)",false],
+["cluster-prefixed identifier","CLUSTERED",false],
+];
+
+it.each(cases)("%s",(_name,sql,want)=>{
+expect(legacyIsPipelineIncompatible(sql)).toBe(want);
+});
 });