diff --git a/apps/cli/src/legacy/commands/db/push/SIDE_EFFECTS.md b/apps/cli/src/legacy/commands/db/push/SIDE_EFFECTS.md index fdf1627eb5..71401c3653 100644 --- a/apps/cli/src/legacy/commands/db/push/SIDE_EFFECTS.md +++ b/apps/cli/src/legacy/commands/db/push/SIDE_EFFECTS.md @@ -27,14 +27,14 @@ before migrations unless `--skip-vault` is set. ## Database Mutations -| Statement | When | -| ------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| `RESET ALL` + migration statements + `INSERT INTO supabase_migrations.schema_migrations(version, name, statements)` | per pending migration (after confirmation); compatible statements use an implicit extended-protocol batch with one final `Sync`, while pipeline-incompatible statements run standalone — see Notes | -| `CREATE SCHEMA/TABLE … supabase_migrations.schema_migrations`, `ALTER TABLE … ADD COLUMN …` | once before applying migrations (idempotent) | -| `roles.sql` statements (no history row) | per `--include-roles` globals file (after confirmation); statements use an implicit extended-protocol batch with one final `Sync` | -| `SELECT id, name FROM vault.secrets …`, `SELECT vault.update_secret(...)`, `SELECT vault.create_secret(...)` | when `[db.vault]` has syncable secrets, migrations are applied, and `--skip-vault` is not set | -| `CREATE TABLE … supabase_migrations.seed_files`, seed statements, `INSERT … seed_files(path, hash) … ON CONFLICT …` | per pending seed file with `--include-seed` (after confirmation); a dirty seed only refreshes the hash | -| `SET SESSION ROLE postgres` | stepped-down sessions only (`cli_login_*`/`supabase_admin`): after each top-level role-reverting statement, at the end of each migration/globals/seed file, and before the history insert and the `seed_files` upsert (CLI-2205, #6236) | +| Statement | When | +| ------------------------------------------------------------------------------------------------------------------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------ | +| `RESET ALL` + migration statements + `INSERT INTO supabase_migrations.schema_migrations(version, name, statements)` | per pending migration (after confirmation); compatible statements use one explicitly transactional extended-protocol batch (`BEGIN` … `COMMIT`) with one final `Sync`, while pipeline-incompatible statements run standalone — see Notes | +| `CREATE SCHEMA/TABLE … supabase_migrations.schema_migrations`, `ALTER TABLE … ADD COLUMN …` | once before applying migrations (idempotent) | +| `roles.sql` statements (no history row) | per `--include-roles` globals file (after confirmation); compatible statements use one explicitly transactional extended-protocol batch (`BEGIN` … `COMMIT`) with one final `Sync`, with the same standalone/sequential exceptions as migrations — see Notes | +| `SELECT id, name FROM vault.secrets …`, `SELECT vault.update_secret(...)`, `SELECT vault.create_secret(...)` | when `[db.vault]` has syncable secrets, migrations are applied, and `--skip-vault` is not set | +| `CREATE TABLE … supabase_migrations.seed_files`, seed statements, `INSERT … seed_files(path, hash) … ON CONFLICT …` | per pending seed file with `--include-seed` (after confirmation); a dirty seed only refreshes the hash | +| `SET SESSION ROLE postgres` | stepped-down sessions only (`cli_login_*`/`supabase_admin`): after each top-level role-reverting statement, at the end of each migration/globals/seed file, and before the history insert and the `seed_files` upsert (CLI-2205, #6236) | ## API Routes @@ -118,11 +118,19 @@ stdout is payload-only. A single `result` object is emitted: load, including decrypted `encrypted:` values. `--skip-vault` leaves them unchanged and does not resolve or decrypt their configured values. - **Pipeline-incompatible statements**: `CREATE [UNIQUE] INDEX CONCURRENTLY`, - `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`, and `CLUSTER` cannot run inside a + `DROP INDEX CONCURRENTLY`, `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`, `CLUSTER`, + `CREATE`/`DROP DATABASE`, `CREATE`/`DROP TABLESPACE`, `REINDEX DATABASE`/`SYSTEM`/`SCHEMA`, + `CREATE`/`DROP SUBSCRIPTION`, `DISCARD ALL`, `ALTER DATABASE … SET TABLESPACE`, and + `ALTER SUBSCRIPTION … REFRESH`/`SET`/`ADD`/`DROP PUBLICATION`, + `ALTER TABLE … DETACH PARTITION … CONCURRENTLY`, + `ALTER TABLE`/`INDEX`/`MATERIALIZED VIEW ALL IN TABLESPACE`, and + `REFRESH MATERIALIZED VIEW CONCURRENTLY` + cannot run inside a transaction block (SQLSTATE 25001). The apply flushes (commits) the open batch, runs the statement standalone outside any transaction, then resumes batching; the history insert stays in the final batch so the migration is recorded only after every - statement succeeds. Atomicity is therefore lost at each flush boundary: statements + statement succeeds. A failed batch's transaction is rolled back (bounded) before its + connection is reused; a rollback that fails or times out discards the connection. Atomicity is therefore lost at each flush boundary: statements committed in an earlier batch are **not** rolled back if a later statement fails, leaving the database partially migrated with **no history row** — a re-run replays the whole file from the top (which may then fail on already-applied statements). diff --git a/apps/cli/src/legacy/commands/db/reset/SIDE_EFFECTS.md b/apps/cli/src/legacy/commands/db/reset/SIDE_EFFECTS.md index 5b2dfb6b5e..54ea002752 100644 --- a/apps/cli/src/legacy/commands/db/reset/SIDE_EFFECTS.md +++ b/apps/cli/src/legacy/commands/db/reset/SIDE_EFFECTS.md @@ -152,7 +152,10 @@ reported as a lost connection (with the driver's own reason and, locally, the hi stack) rather than against a statement that never ran. Once any part of the batch has been written, and for the pipeline-incompatible statements the same loop runs on their own (`CREATE INDEX CONCURRENTLY`, `VACUUM`, ...), a failure still reports as `At statement: N` with the statement -echoed, because those may genuinely have reached the server. +echoed, because those may genuinely have reached the server. The one exception is a failure of the +batch's own transaction wrapper — a rejected `BEGIN`, or a deferred constraint surfacing at +`COMMIT` — which reports the phase-labeled driver message with no statement context, since no +caller statement is to blame. ## Exit Codes diff --git a/apps/cli/src/legacy/commands/migration/up/SIDE_EFFECTS.md b/apps/cli/src/legacy/commands/migration/up/SIDE_EFFECTS.md index 8a700a4c12..932a2bdc1e 100644 --- a/apps/cli/src/legacy/commands/migration/up/SIDE_EFFECTS.md +++ b/apps/cli/src/legacy/commands/migration/up/SIDE_EFFECTS.md @@ -65,10 +65,18 @@ Same structured `applied` result delivered as an NDJSON `result` event. a non-linked target). - `--include-all` applies all migrations not found on the remote history table. - Pipeline-incompatible statements (`CREATE [UNIQUE] INDEX CONCURRENTLY`, - `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`, `CLUSTER`) run standalone outside + `DROP INDEX CONCURRENTLY`, `REINDEX … CONCURRENTLY`, `VACUUM`, `ALTER SYSTEM`, + `CLUSTER`, `CREATE`/`DROP DATABASE`, `CREATE`/`DROP TABLESPACE`, + `REINDEX DATABASE`/`SYSTEM`/`SCHEMA`, `CREATE`/`DROP SUBSCRIPTION`, `DISCARD ALL`, + `ALTER DATABASE … SET TABLESPACE`, + `ALTER SUBSCRIPTION … REFRESH`/`SET`/`ADD`/`DROP PUBLICATION`, + `ALTER TABLE … DETACH PARTITION … CONCURRENTLY`, + `ALTER TABLE`/`INDEX`/`MATERIALIZED VIEW ALL IN TABLESPACE`, and + `REFRESH MATERIALIZED VIEW CONCURRENTLY`) run standalone outside the migration's transaction batch — they fail with SQLSTATE 25001 inside one. The history insert stays in the final batch, so a mid-file failure leaves earlier, already-committed batches applied with **no history row**; a re-run replays the file - from the top. Prefer idempotent forms (`… IF NOT EXISTS`) for such statements. + from the top. A failed batch's transaction is rolled back (bounded) before its + connection is reused. Prefer idempotent forms (`… IF NOT EXISTS`) for such statements. Intentional fix for supabase/cli#5139, adopted into TS in PR supabase/cli#5671 (landed on develop as `b48fad60`). diff --git a/apps/cli/src/legacy/shared/legacy-db-connection.errors.ts b/apps/cli/src/legacy/shared/legacy-db-connection.errors.ts index 99fda4cef2..72b683d183 100644 --- a/apps/cli/src/legacy/shared/legacy-db-connection.errors.ts +++ b/apps/cli/src/legacy/shared/legacy-db-connection.errors.ts @@ -36,6 +36,12 @@ export class LegacyDbExecError extends Data.TaggedError("LegacyDbExecError")<{ * 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 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 e19ce440b8..76fa8efdad 100644 --- a/apps/cli/src/legacy/shared/legacy-db-connection.service.ts +++ b/apps/cli/src/legacy/shared/legacy-db-connection.service.ts @@ -108,9 +108,12 @@ export interface LegacyDbSession { /** Run a single SQL statement, ignoring any returned rows. */ readonly exec: (sql: string) => Effect.Effect; /** - * Run statements as one extended-protocol batch with a single final Sync. - * On failure, {@link LegacyDbExecError.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 LegacyDbExecError.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 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 f872fe031f..9be4937c55 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 @@ -11,7 +11,7 @@ import { describe, expect, it } from "@effect/vitest"; import { Duration, Effect } from "effect"; import { LEGACY_SUGGEST_ENV_VAR, LEGACY_SUGGEST_LOCAL_STACK } from "./legacy-connect-errors.ts"; -import type { LegacyDbConnectError, LegacyDbExecError } from "./legacy-db-connection.errors.ts"; +import { type LegacyDbConnectError, LegacyDbExecError } from "./legacy-db-connection.errors.ts"; import { type LegacyDbSession, type LegacyPgConnInput, @@ -166,6 +166,7 @@ interface FakeBatchServerState { readonly frameTypes: Array; readonly statements: Array; readonly params: Array>; + readonly simpleQueries: Array; syncs: number; } @@ -177,8 +178,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; @@ -191,9 +196,11 @@ const fakeBatchServer = ( frameTypes: [], statements: [], params: [], + simpleQueries: [], syncs: 0, }; const sockets: Array = []; + let destroyedOnExecute = false; const server = net.createServer((socket) => { sockets.push(socket); let sawStartup = false; @@ -223,6 +230,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; } @@ -254,6 +268,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; @@ -554,9 +573,9 @@ describe("legacyDbConnectionSqlPgLayer 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([ @@ -569,11 +588,14 @@ describe("legacyDbConnectionSqlPgLayer 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([ + [], [], [], [ @@ -581,6 +603,7 @@ describe("legacyDbConnectionSqlPgLayer extended batches", () => { "name\\two", '{"plain","quote\\\"","slash\\\\","comma,","{brace}","line\nbreak","NULL",""}', ], + [], ]); expect(server.state.frameTypes).toEqual([ "P", @@ -595,15 +618,24 @@ describe("legacyDbConnectionSqlPgLayer 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( @@ -624,7 +656,7 @@ describe("legacyDbConnectionSqlPgLayer 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" }]) @@ -662,6 +694,72 @@ describe("legacyDbConnectionSqlPgLayer 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. @@ -678,6 +776,35 @@ describe("legacyDbConnectionSqlPgLayer extended batches", () => { ); expect(error._tag).toBe("LegacyDbExecError"); 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(LegacyDbExecError); + expect(asBatchExecError(error).message).toContain("Connection terminated unexpectedly"); + expect(asBatchExecError(error)).toMatchObject({ + statementIndex: 0, + transactionPhase: "begin", + }); yield* session.execBatch([{ sql: "SELECT 3" }]); }), ); 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 a0cd4c1fbe..d3c16fee4d 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 @@ -2,7 +2,7 @@ import { readFileSync } from "node:fs"; import * as net from "node:net"; import type { ConnectionOptions } from "node:tls"; import { PgClient } from "@effect/sql-pg"; -import { Cause, Duration, Effect, Exit, Layer, Scope } from "effect"; +import { Cause, Duration, Effect, Exit, Layer, Option, Scope } from "effect"; import * as Reactivity from "effect/unstable/reactivity/Reactivity"; import { ConnectionError, SqlError } from "effect/unstable/sql/SqlError"; // `pg` is also `@effect/sql-pg`'s transitive driver; we depend on it directly for @@ -109,6 +109,12 @@ function needsRoleStepDown(user: string): boolean { const LEGACY_TERMINAL_SQLSTATES = new Set(["28P01", "3D000", "42501"]); const LEGACY_TLS_GATED_SQLSTATE = "28000"; +// Class 08 (connection exception) plus the operator-intervention terminations that +// close the session; 57014 (query_canceled) stays a statement failure. +const LEGACY_SESSION_ENDING_SQLSTATES = new Set(["57P01", "57P02", "57P03", "57P04", "57P05"]); +const legacyIsConnectionEndingSqlState = (code: string): boolean => + code.startsWith("08") || LEGACY_SESSION_ENDING_SQLSTATES.has(code); + /** * Whether a failed connection attempt should terminate the multi-host fallback * chain instead of falling through to the next host. Mirrors pgconn's @@ -224,7 +230,14 @@ const LEGACY_DB_KEEPALIVE_IDLE_MILLIS = 300_000; */ export function legacyBatchFailureError( error: Error, - batch: { readonly completed: number; readonly outcome: LegacyBatchOutcome } | undefined, + batch: + | { + readonly completed: number; + readonly outcome: LegacyBatchOutcome; + readonly began?: boolean; + readonly atCommit?: boolean; + } + | undefined, isLocal: boolean, ): LegacyDbExecError | LegacyDbConnectError { if (batch === undefined || batch.outcome === "unsent") { @@ -236,12 +249,34 @@ export function legacyBatchFailureError( }); } const mapped = legacyToExecError(error); + // The phase marker and the relabel are separate: whenever BEGIN or COMMIT was + // the statement in flight, none of the caller's statements failed at + // `statementIndex`, so the phase is always recorded and formatters must not + // blame one. The message is only relabeled when the server rejected the + // wrapper itself — a lost connection (including a server-initiated + // termination) keeps its own reason. Gated on SQLSTATE class, never the + // severity string, which arrives localized (e.g. "FEHLER"). + const server = legacyExtractPgServerError(error); + const statementFailure = server !== undefined && !legacyIsConnectionEndingSqlState(server.code); + const atBegin = batch.outcome === "submitted" && batch.began === false; + const atCommit = batch.outcome === "submitted" && batch.atCommit === true; + const transactionPhase: "begin" | "commit" | undefined = atBegin + ? "begin" + : atCommit + ? "commit" + : undefined; return new LegacyDbExecError({ - message: mapped.message, + message: + atBegin && statementFailure + ? `failed to begin the batch transaction: ${mapped.message}` + : atCommit && statementFailure + ? `failed to commit the batch transaction: ${mapped.message}` + : mapped.message, code: mapped.code, detail: mapped.detail, position: mapped.position, statementIndex: batch.completed, + ...(transactionPhase !== undefined ? { transactionPhase } : {}), }); } @@ -251,18 +286,24 @@ export function legacyBatchFailureError( * 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. + * a fresh step-down on a single-connection pool. The keep is conditional on `rolledBack` — + * a failed submitted batch whose rollback failed or timed out is discarded. Recovering + * from a socket that died after the write is additionally backstopped by 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, + rolledBack: boolean, ): boolean { return ( (batch !== undefined && batch.outcome !== "submitted") || - (Exit.isFailure(exit) && (Cause.hasInterrupts(exit.cause) || Cause.hasDies(exit.cause))) + (Exit.isFailure(exit) && + (Cause.hasInterrupts(exit.cause) || + Cause.hasDies(exit.cause) || + (batch?.outcome === "submitted" && !rolledBack))) ); } @@ -282,6 +323,12 @@ export class LegacyPgBatchQuery implements Pg.Submittable { callback: (error: Error | undefined) => void; completed = 0; outcome: LegacyBatchOutcome = "unsent"; + began = false; + + // An error arriving once every caller statement completed can only be COMMIT's. + get atCommit(): boolean { + return this.began && this.completed >= this.statements.length; + } constructor( statements: ReadonlyArray, @@ -301,7 +348,12 @@ export class LegacyPgBatchQuery implements Pg.Submittable { let started = false; connection.stream.cork?.(); try { - for (const { sql, params } of this.statements) { + // A bare pipeline is not a transaction block (supabase/cli#6347). + for (const { sql, params } of [ + { sql: "BEGIN", params: [] }, + ...this.statements, + { sql: "COMMIT", params: [] }, + ]) { started = true; connection.parse({ name: "", text: sql, types: [] }, true); connection.bind({ portal: "", statement: "", values: [...params] }, true); @@ -332,11 +384,22 @@ export class LegacyPgBatchQuery implements Pg.Submittable { handlePortalSuspended(): void {} handleCommandComplete(): void { - this.completed += 1; + this.recordCompletion(); } handleEmptyQuery(): void { - this.completed += 1; + this.recordCompletion(); + } + + // BEGIN completes first and COMMIT only after every statement; neither counts. + private recordCompletion(): void { + if (!this.began) { + this.began = true; + return; + } + if (this.completed < this.statements.length) { + this.completed += 1; + } } handleCopyInResponse(connection: Pg.Connection): void { @@ -1084,44 +1147,70 @@ const connect = ( const execBatch = (statements: ReadonlyArray) => { if (statements.length === 0) return Effect.void; - let batchQuery: LegacyPgBatchQuery | undefined; - return Effect.acquireUseRelease( - Effect.interruptible(acquireBatchClient), - (activeClient) => { - const onConnectionError = () => {}; - activeClient.on("error", onConnectionError); - return Effect.callback((resume) => { - let done = false; - const finish = (error: Error | undefined) => { - if (done) return; - done = true; - if (error === undefined) { - resume(Effect.void); - return; + // Suspended so each evaluation owns fresh batch/rollback state. + return Effect.suspend(() => { + let batchQuery: LegacyPgBatchQuery | undefined; + let rolledBack = false; + // Spans the whole checkout: an unlistened 'error' kills the process (see + // acquireRawClient) and pg-pool detaches its own handler while checked out. + const onConnectionError = () => {}; + return Effect.acquireUseRelease( + Effect.interruptible(acquireBatchClient).pipe( + Effect.tap((activeClient) => + Effect.sync(() => activeClient.on("error", onConnectionError)), + ), + ), + (activeClient) => + Effect.callback((resume) => { + let done = false; + const finish = (error: Error | undefined) => { + if (done) return; + done = true; + if (error === undefined) { + resume(Effect.void); + return; + } + resume(Effect.fail(legacyBatchFailureError(error, batchQuery, options.isLocal))); + }; + batchQuery = new LegacyPgBatchQuery(statements, finish); + try { + activeClient.query(batchQuery); + } catch (error) { + finish(error instanceof Error ? error : new Error(String(error))); } - resume(Effect.fail(legacyBatchFailureError(error, batchQuery, options.isLocal))); - }; - batchQuery = new LegacyPgBatchQuery(statements, finish); - try { - activeClient.query(batchQuery); - } catch (error) { - finish(error instanceof Error ? error : new Error(String(error))); - } - return Effect.sync(() => { - done = true; - }); - }).pipe( - Effect.ensuring( - Effect.sync(() => activeClient.removeListener("error", onConnectionError)), + return Effect.sync(() => { + done = true; + }); + }).pipe( + // Roll a written batch's aborted transaction back while still + // interruptible; a rollback that fails or times out leaves the client + // to the discard below instead of returning it aborted (25P02). The + // rollback's own failure is consumed as that discard policy — it must + // never supplant the batch error this tap is observing. + Effect.tapError(() => + Effect.suspend(() => { + if (batchQuery?.outcome !== "submitted") return Effect.void; + return Effect.tryPromise(() => activeClient.query("ROLLBACK")).pipe( + Effect.match({ onFailure: () => false, onSuccess: () => true }), + Effect.timeoutOption(1000), + Effect.map((completed) => { + rolledBack = Option.getOrElse(completed, () => false); + }), + ); + }), + ), ), - ); - }, - (activeClient, exit) => - Effect.sync(() => { - const discard = legacyShouldDiscardBatchClient(batchQuery, exit); - activeClient.release(discard ? new Error("batch connection discarded") : undefined); - }), - ); + (activeClient, exit) => + Effect.sync(() => { + const discard = legacyShouldDiscardBatchClient(batchQuery, exit, rolledBack); + try { + activeClient.release(discard ? new Error("batch connection discarded") : undefined); + } finally { + activeClient.removeListener("error", onConnectionError); + } + }), + ); + }); }; const session: LegacyDbSession = { 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 6211b4050c..495b1bde1c 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 @@ -5,6 +5,7 @@ import type * as Pg from "pg"; import { describe, expect, it } from "vitest"; import { ErrorActionabilityId } from "../../shared/telemetry/error-actionability.ts"; +import { LegacyDbExecError } from "./legacy-db-connection.errors.ts"; import { LEGACY_SUGGEST_LOCAL_STACK } from "./legacy-connect-errors.ts"; import { legacyAcquireProbedPool, @@ -608,7 +609,9 @@ describe("LegacyPgBatchQuery.submit", () => { frames, connection: { stream, - parse: record("parse"), + parse: (query: { text: string }) => { + frames.push(`parse(${query.text})`); + }, bind: record("bind"), describe: record("describe"), execute: record("execute"), @@ -638,10 +641,26 @@ describe("LegacyPgBatchQuery.submit", () => { "the connection's socket became unwritable while the batch was flushing", ); expect(batch.outcome).toBe("unsent"); - expect(frames).toEqual(["cork", "parse", "bind", "describe", "execute", "sync", "uncork"]); + expect(frames).toEqual([ + "cork", + "parse(BEGIN)", + "bind", + "describe", + "execute", + "parse(select 1)", + "bind", + "describe", + "execute", + "parse(COMMIT)", + "bind", + "describe", + "execute", + "sync", + "uncork", + ]); }); - it("writes parse/bind/describe/execute per statement and one sync while writable", () => { + it("brackets the statements in BEGIN/COMMIT and writes one sync while writable", () => { const { connection, frames } = fakeConnection(true); const batch = new LegacyPgBatchQuery([{ sql: "select 1" }, { sql: "select 2" }], () => {}); @@ -649,11 +668,19 @@ describe("LegacyPgBatchQuery.submit", () => { expect(frames).toEqual([ "cork", - "parse", + "parse(BEGIN)", + "bind", + "describe", + "execute", + "parse(select 1)", + "bind", + "describe", + "execute", + "parse(select 2)", "bind", "describe", "execute", - "parse", + "parse(COMMIT)", "bind", "describe", "execute", @@ -662,6 +689,18 @@ describe("LegacyPgBatchQuery.submit", () => { ]); expect(batch.outcome).toBe("submitted"); }); + + it("counts neither BEGIN's nor COMMIT's completion toward the statement index", () => { + const batch = new LegacyPgBatchQuery([{ sql: "select 1" }, { sql: "select 2" }], () => {}); + + batch.handleCommandComplete(); + expect(batch.completed).toBe(0); + batch.handleCommandComplete(); + batch.handleEmptyQuery(); + expect(batch.completed).toBe(2); + batch.handleCommandComplete(); + expect(batch.completed).toBe(2); + }); }); describe("legacyBatchFailureError", () => { @@ -737,6 +776,124 @@ describe("legacyBatchFailureError", () => { expect(error).toMatchObject({ message: "Error: serialization blew up", statementIndex: 0 }); }); + it("names the transaction start when the server rejected the batch before BEGIN completed", () => { + const beginRejected = new SqlError({ + reason: new SqlSyntaxError({ + cause: Object.assign(new Error("canceling statement due to statement timeout"), { + severity: "ERROR", + code: "57014", + }), + message: "Failed to execute statement", + operation: "execute", + }), + }); + const error = legacyBatchFailureError( + beginRejected, + { completed: 0, outcome: "submitted", began: false }, + true, + ); + + expect(error).toBeInstanceOf(LegacyDbExecError); + expect(error).toMatchObject({ + message: + "failed to begin the batch transaction: " + + "ERROR: canceling statement due to statement timeout (SQLSTATE 57014)", + statementIndex: 0, + transactionPhase: "begin", + }); + + const lost = legacyBatchFailureError( + new Error("Connection terminated unexpectedly"), + { completed: 0, outcome: "submitted", began: false }, + true, + ); + expect(lost).toBeInstanceOf(LegacyDbExecError); + // A connection lost at BEGIN keeps its own reason, but still marks the phase: + // no caller statement ran, so nothing downstream may render `At statement: 0`. + expect(lost).toMatchObject({ + message: "Error: Connection terminated unexpectedly", + statementIndex: 0, + transactionPhase: "begin", + }); + + const terminated = legacyBatchFailureError( + new SqlError({ + reason: new SqlSyntaxError({ + cause: Object.assign(new Error("terminating connection due to idle-session timeout"), { + severity: "FATAL", + code: "57P05", + }), + message: "Failed to execute statement", + operation: "execute", + }), + }), + { completed: 0, outcome: "submitted", began: false }, + true, + ); + expect(terminated).toBeInstanceOf(LegacyDbExecError); + expect(terminated).toMatchObject({ + message: "FATAL: terminating connection due to idle-session timeout (SQLSTATE 57P05)", + transactionPhase: "begin", + }); + + const poisoned = legacyBatchFailureError( + beginRejected, + { completed: 0, outcome: "poisoned", began: false }, + true, + ); + expect(poisoned).toBeInstanceOf(LegacyDbExecError); + expect(poisoned.message).toBe( + "ERROR: canceling statement due to statement timeout (SQLSTATE 57014)", + ); + // A poisoned batch never reached the server, so it stays on the statement path. + expect(poisoned).not.toHaveProperty("transactionPhase"); + }); + + it("names the transaction commit when a deferred failure lands on COMMIT", () => { + const deferred = new SqlError({ + reason: new SqlSyntaxError({ + cause: Object.assign(new Error("deferred constraint failed"), { + severity: "ERROR", + code: "23514", + }), + message: "Failed to execute statement", + operation: "execute", + }), + }); + const error = legacyBatchFailureError( + deferred, + { completed: 2, outcome: "submitted", began: true, atCommit: true }, + true, + ); + expect(error).toBeInstanceOf(LegacyDbExecError); + expect(error).toMatchObject({ + message: + "failed to commit the batch transaction: ERROR: deferred constraint failed (SQLSTATE 23514)", + statementIndex: 2, + transactionPhase: "commit", + }); + + const dropped = legacyBatchFailureError( + new SqlError({ + reason: new SqlSyntaxError({ + cause: Object.assign(new Error("terminating connection: database dropped"), { + severity: "FATAL", + code: "57P04", + }), + message: "Failed to execute statement", + operation: "execute", + }), + }), + { completed: 2, outcome: "submitted", began: true, atCommit: true }, + true, + ); + expect(dropped).toBeInstanceOf(LegacyDbExecError); + expect(dropped).toMatchObject({ + message: "FATAL: terminating connection: database dropped (SQLSTATE 57P04)", + transactionPhase: "commit", + }); + }); + it("keeps server-error mapping and the completed count for a statement failure", () => { const error = legacyBatchFailureError( new SqlError({ @@ -768,27 +925,42 @@ describe("legacyBatchFailureError", () => { describe("legacyShouldDiscardBatchClient", () => { it("discards a client whose batch never reached the wire", () => { - expect(legacyShouldDiscardBatchClient({ outcome: "unsent" }, Exit.succeed(undefined))).toBe( - true, - ); + expect( + legacyShouldDiscardBatchClient({ outcome: "unsent" }, Exit.succeed(undefined), false), + ).toBe(true); }); - it("returns a client to the pool once its batch was written, error or not", () => { - expect(legacyShouldDiscardBatchClient({ outcome: "submitted" }, Exit.succeed(undefined))).toBe( - false, - ); + it("keeps a written batch's client on success or once its failure rolled back", () => { + expect( + legacyShouldDiscardBatchClient({ outcome: "submitted" }, Exit.succeed(undefined), false), + ).toBe(false); expect( legacyShouldDiscardBatchClient( { outcome: "submitted" }, Exit.fail(new Error("server said no")), + true, ), ).toBe(false); }); + it("discards a written batch's client when its failure was not rolled back", () => { + expect( + legacyShouldDiscardBatchClient( + { outcome: "submitted" }, + Exit.fail(new Error("server said no")), + false, + ), + ).toBe(true); + }); + it("discards a client whose batch was interrupted or died mid-flight", () => { - expect(legacyShouldDiscardBatchClient({ outcome: "submitted" }, Exit.interrupt(1))).toBe(true); - expect(legacyShouldDiscardBatchClient({ outcome: "submitted" }, Exit.die("boom"))).toBe(true); + expect(legacyShouldDiscardBatchClient({ outcome: "submitted" }, Exit.interrupt(1), true)).toBe( + true, + ); + expect(legacyShouldDiscardBatchClient({ outcome: "submitted" }, Exit.die("boom"), true)).toBe( + true, + ); // Interrupted before the batch was even constructed: no batch, still discard. - expect(legacyShouldDiscardBatchClient(undefined, Exit.interrupt(1))).toBe(true); + expect(legacyShouldDiscardBatchClient(undefined, Exit.interrupt(1), false)).toBe(true); }); }); diff --git a/apps/cli/src/legacy/shared/legacy-migration-apply.ts b/apps/cli/src/legacy/shared/legacy-migration-apply.ts index 806973a2e3..0b5e99c285 100644 --- a/apps/cli/src/legacy/shared/legacy-migration-apply.ts +++ b/apps/cli/src/legacy/shared/legacy-migration-apply.ts @@ -61,9 +61,24 @@ const BOM_CODE_POINT = 0xfeff; const CREATE_INDEX_CONCURRENTLY_PATTERN = /^CREATE\s+(?:UNIQUE\s+)?INDEX\s+CONCURRENTLY(?:\s|$)/u; const DROP_INDEX_CONCURRENTLY_PATTERN = /^DROP\s+INDEX\s+CONCURRENTLY(?:\s|$)/u; const REINDEX_CONCURRENTLY_PATTERN = /^REINDEX(?:\s|\().*\sCONCURRENTLY(?:\s|$)/u; +const REINDEX_OPTION_CONCURRENTLY_PATTERN = /^REINDEX\s*\([^)]*\bCONCURRENTLY\b[^)]*\)/u; const VACUUM_PATTERN = /^VACUUM(?:\s|\(|$)/u; const ALTER_SYSTEM_PATTERN = /^ALTER\s+SYSTEM(?:\s|$)/u; const CLUSTER_PATTERN = /^CLUSTER(?:\s|$)/u; +const DATABASE_DDL_PATTERN = /^(?:CREATE|DROP)\s+DATABASE(?:\s|$)/u; +const TABLESPACE_DDL_PATTERN = /^(?:CREATE|DROP)\s+TABLESPACE(?:\s|$)/u; +const REINDEX_DATABASE_PATTERN = /^REINDEX(?:\s*\([^)]*\))?\s+(?:DATABASE|SYSTEM|SCHEMA)(?:\s|$)/u; +const SUBSCRIPTION_DDL_PATTERN = /^(?:CREATE|DROP)\s+SUBSCRIPTION(?:\s|$)/u; +const DISCARD_ALL_PATTERN = /^DISCARD\s+ALL(?:\s|$)/u; +const ALTER_DATABASE_TABLESPACE_PATTERN = /^ALTER\s+DATABASE\s[\s\S]*\sSET\s+TABLESPACE(?:\s|$)/u; +const ALTER_SUBSCRIPTION_REFRESH_PATTERN = + /^ALTER\s+SUBSCRIPTION\s[\s\S]*\s(?:REFRESH\s+PUBLICATION|(?:SET|ADD|DROP)\s+PUBLICATION)(?:\s|$)/u; +const ALL_IN_TABLESPACE_PATTERN = + /^ALTER\s+(?:TABLE|INDEX|MATERIALIZED\s+VIEW)\s+ALL\s+IN\s+TABLESPACE(?:\s|$)/u; +const DETACH_PARTITION_PATTERN = + /^ALTER\s+TABLE\s[\s\S]*\sDETACH\s+PARTITION\s[\s\S]*\sCONCURRENTLY(?:\s|$)/u; +const REFRESH_MATERIALIZED_VIEW_CONCURRENTLY_PATTERN = + /^REFRESH\s+MATERIALIZED\s+VIEW\s+CONCURRENTLY(?:\s|$)/u; const TRANSACTION_CONTROL_PATTERN = /^(?:BEGIN|START\s+TRANSACTION|COMMIT|END|ABORT|PREPARE\s+TRANSACTION)(?:\s|$)/u; @@ -97,10 +112,26 @@ const legacyTrimLeadingSqlComments = (sql: string): string => { /** * Whether a migration statement cannot run inside a transaction block — `CREATE * [UNIQUE] INDEX CONCURRENTLY`, `DROP INDEX CONCURRENTLY`, `REINDEX … CONCURRENTLY`, - * `VACUUM`, `ALTER SYSTEM`, `CLUSTER`. Such statements fail with SQLSTATE 25001 - * inside the implicit transaction + * `VACUUM`, `ALTER SYSTEM`, `CLUSTER`, `CREATE`/`DROP DATABASE`, + * `CREATE`/`DROP TABLESPACE`, `REINDEX DATABASE`/`SYSTEM`/`SCHEMA`, + * `CREATE`/`DROP SUBSCRIPTION`, `DISCARD ALL`, + * `ALTER DATABASE … SET TABLESPACE`, + * `ALTER SUBSCRIPTION … REFRESH`/`SET`/`ADD`/`DROP PUBLICATION`, + * `ALTER TABLE … DETACH PARTITION … CONCURRENTLY`, + * `ALTER TABLE`/`INDEX`/`MATERIALIZED VIEW ALL IN TABLESPACE`, + * `REFRESH MATERIALIZED VIEW CONCURRENTLY`. Such statements (in + * their default forms) fail with + * SQLSTATE 25001 inside the transaction * created by a migration batch, so `execMigrationBatch` runs them standalone. - * Port of `isPipelineIncompatible` (`pkg/migration/file.go`, supabase/cli#5156). + * Port of `isPipelineIncompatible` (`pkg/migration/file.go`, supabase/cli#5156), + * extended with the remaining statement kinds PostgreSQL refuses inside the + * explicit transaction the batch runs in since supabase/cli#6347. + * + * Deliberately loose, keyword-anchored matching — never identifier-aware: a match + * inside a comment or literal over-routes the statement standalone, which is always + * valid SQL placement and at worst adds a documented flush boundary, while an + * under-match is a hard SQLSTATE 25001 failure. Do not tighten these into + * identifier parsing. */ export const legacyIsPipelineIncompatible = (sql: string): boolean => { const upper = legacyTrimLeadingSqlComments(sql).toUpperCase(); @@ -108,9 +139,20 @@ export const legacyIsPipelineIncompatible = (sql: string): boolean => { CREATE_INDEX_CONCURRENTLY_PATTERN.test(upper) || DROP_INDEX_CONCURRENTLY_PATTERN.test(upper) || REINDEX_CONCURRENTLY_PATTERN.test(upper) || + REINDEX_OPTION_CONCURRENTLY_PATTERN.test(upper) || VACUUM_PATTERN.test(upper) || ALTER_SYSTEM_PATTERN.test(upper) || - CLUSTER_PATTERN.test(upper) + CLUSTER_PATTERN.test(upper) || + DATABASE_DDL_PATTERN.test(upper) || + TABLESPACE_DDL_PATTERN.test(upper) || + REINDEX_DATABASE_PATTERN.test(upper) || + SUBSCRIPTION_DDL_PATTERN.test(upper) || + DISCARD_ALL_PATTERN.test(upper) || + ALTER_DATABASE_TABLESPACE_PATTERN.test(upper) || + ALTER_SUBSCRIPTION_REFRESH_PATTERN.test(upper) || + DETACH_PARTITION_PATTERN.test(upper) || + ALL_IN_TABLESPACE_PATTERN.test(upper) || + REFRESH_MATERIALIZED_VIEW_CONCURRENTLY_PATTERN.test(upper) ); }; @@ -531,8 +573,8 @@ const formattedExecBatchDbError = (error: unknown): LegacyDbExecError | undefine /** * Runs a single migration/seed file's statements (plus the optional history insert). - * Statements run inside an implicitly transactional extended-protocol batch, - * except pipeline-incompatible ones + * Statements run inside an explicitly transactional extended-protocol batch + * (supabase/cli#6347), except pipeline-incompatible ones * (`legacyIsPipelineIncompatible` — `CREATE INDEX CONCURRENTLY`, `VACUUM`, …) which * cannot run in a transaction block: the open batch is flushed (committed), the * statement runs standalone, then batching resumes (supabase/cli#5156). The history @@ -706,11 +748,14 @@ const execMigrationBatch = ( // (Go threads the same counter through `ExecBatch`). let pending: Array = []; let executed = 0; + let standaloneRestored = false; const flushBatch = (final: boolean) => Effect.gen(function* () { const recordVersion = final && version.length > 0; - const trailingRestore = final ? restoreRole : undefined; + // A standalone statement that just restored the role needs no trailing repeat. + const trailingRestore = + final && !(pending.length === 0 && standaloneRestored) ? restoreRole : undefined; if (pending.length === 0 && !recordVersion && trailingRestore === undefined) return; const batchStatements = pending; const operations: Array = []; @@ -754,6 +799,16 @@ const execMigrationBatch = ( // `statementIndex` is set by every batch failure the driver raises; a // session that omits it can only have failed before the first statement. const raw = cause.statementIndex ?? 0; + // A wrapper (BEGIN/COMMIT) failure, or one past every operation + // (e.g. a deferred constraint), blames no statement: keep the + // driver's phase-labeled message without an `At statement` tail. + if (cause.transactionPhase !== undefined || raw >= operations.length) { + const msg = [legacyErrorMessage(cause)]; + if (cause.detail !== undefined && cause.detail.length > 0) { + msg.push(cause.detail); + } + return formattedExecBatchFailure(msg.join("\n"), cause); + } const globalIndex = base + raw - (injectedBefore[raw] ?? injected); return legacyFormatExecBatchError( cause, @@ -775,9 +830,19 @@ const execMigrationBatch = ( yield* session .exec(statement) .pipe(Effect.mapError((cause) => legacyFormatExecBatchError(cause, index, statement))); + standaloneRestored = false; + if (restoreRole !== undefined && legacyRevertsToLoginRole(statement)) { + yield* session + .exec(restoreRole) + .pipe( + Effect.mapError((cause) => legacyFormatExecBatchError(cause, index, restoreRole)), + ); + standaloneRestored = true; + } executed += 1; } else { pending.push(statement); + standaloneRestored = false; } } yield* flushBatch(true); 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 14b75f1d73..14bf5693eb 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 @@ -31,6 +31,7 @@ class FakeExecError extends Data.TaggedError("LegacyDbExecError")<{ readonly detail?: string; readonly position?: number; readonly statementIndex?: number; + readonly transactionPhase?: "begin" | "commit"; }> { get [ErrorActionabilityId](): CliErrorActionabilityDeclaration { return actionability.dbFinding; @@ -41,7 +42,13 @@ function fakeSession( opts: { failOn?: string; failAfterBatch?: boolean; - failWith?: { message: string; code?: string; detail?: string; position?: number }; + failWith?: { + message: string; + code?: string; + detail?: string; + position?: number; + transactionPhase?: "begin" | "commit"; + }; restoreRoleSql?: string; batchConnectionLost?: string; } = {}, @@ -284,7 +291,58 @@ describe("legacyApplyMigrationFile", () => { ); }); - it.effect("defaults a deferred batch failure to the migration history statement", () => { + it.effect("reports a begin failure without blaming the first statement", () => { + const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); + const file = join(dir, "20240101120000_begin.sql"); + writeFileSync(file, "SELECT 1;"); + const { session } = fakeSession({ + failOn: "SELECT 1", + failWith: { + message: "failed to begin the batch transaction: ERROR: canceling statement", + transactionPhase: "begin", + }, + }); + return run(session, file).pipe( + Effect.flip, + Effect.tap((error) => + Effect.sync(() => { + expect(error.message).toContain("failed to begin the batch transaction"); + expect(error.message).not.toContain("At statement"); + expect(error.message).not.toContain("SELECT 1"); + }), + ), + Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true }))), + ); + }); + + it.effect("reports a connection lost at BEGIN without blaming the first statement", () => { + // The driver marks the begin phase without relabeling the message when the + // connection died before BEGIN completed; the caller's statement never ran, + // so the formatter must surface the loss verbatim with no statement echo. + const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); + const file = join(dir, "20240101120000_begin_lost.sql"); + writeFileSync(file, "SELECT 1;"); + const { session } = fakeSession({ + failOn: "SELECT 1", + failWith: { + message: "Error: Connection terminated unexpectedly", + transactionPhase: "begin", + }, + }); + return run(session, file).pipe( + Effect.flip, + Effect.tap((error) => + Effect.sync(() => { + expect(error.message).toContain("Connection terminated unexpectedly"); + expect(error.message).not.toContain("At statement"); + expect(error.message).not.toContain("SELECT 1"); + }), + ), + Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true }))), + ); + }); + + it.effect("reports a deferred commit failure without blaming a statement", () => { const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); const file = join(dir, "20240101120000_deferred.sql"); writeFileSync(file, "SELECT 1;"); @@ -293,8 +351,8 @@ describe("legacyApplyMigrationFile", () => { Effect.flip, Effect.tap((error) => Effect.sync(() => { - expect(error.message).toContain("At statement: 2"); - expect(error.message).toContain("INSERT INTO supabase_migrations.schema_migrations"); + expect(error.message).not.toContain("At statement"); + expect(error.message).not.toContain("INSERT INTO supabase_migrations.schema_migrations"); }), ), Effect.ensuring(Effect.sync(() => rmSync(dir, { recursive: true, force: true }))), @@ -656,10 +714,9 @@ describe("legacyApplyMigrationFile", () => { ); }); - it.effect("keeps the deferred-failure index when a restore op is appended", () => { - // Mirrors "defaults a deferred batch failure to the migration history - // statement": the restore op between the statements and the insert must not - // shift the deferred (post-Sync) index either. + it.effect("reports a deferred commit failure cleanly when a restore op is appended", () => { + // The restore op between the statements and the insert must not resurrect a + // statement tail for a commit-phase failure. const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); const file = join(dir, "20240101120000_deferred.sql"); writeFileSync(file, "SELECT 1;"); @@ -671,8 +728,8 @@ describe("legacyApplyMigrationFile", () => { Effect.flip, Effect.tap((error) => Effect.sync(() => { - expect(error.message).toContain("At statement: 2"); - expect(error.message).toContain("INSERT INTO supabase_migrations.schema_migrations"); + expect(error.message).not.toContain("At statement"); + expect(error.message).not.toContain("INSERT INTO supabase_migrations.schema_migrations"); rmSync(dir, { recursive: true, force: true }); }), ), @@ -808,7 +865,7 @@ describe("legacyApplyMigrationFile", () => { ); }); - it.effect("keeps the deferred-failure index when mid-file restores were injected", () => { + it.effect("reports a deferred commit failure cleanly with mid-file restores injected", () => { // The `injectedBefore[raw] ?? injected` fallback only matters when the // deferred (post-Sync) index lands past the ops array AND injections exist. const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); @@ -822,8 +879,8 @@ describe("legacyApplyMigrationFile", () => { Effect.flip, Effect.tap((error) => Effect.sync(() => { - expect(error.message).toContain("At statement: 4"); - expect(error.message).toContain("INSERT INTO supabase_migrations.schema_migrations"); + expect(error.message).not.toContain("At statement"); + expect(error.message).not.toContain("INSERT INTO supabase_migrations.schema_migrations"); rmSync(dir, { recursive: true, force: true }); }), ), @@ -854,6 +911,49 @@ describe("legacyApplyMigrationFile", () => { ); }); + it.effect("re-asserts postgres right after a standalone role-reverting statement", () => { + const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); + const file = join(dir, "20240101120000_discard.sql"); + writeFileSync(file, "select 1;\nDISCARD ALL;\nselect 2;"); + const { session, calls } = fakeSession({ restoreRoleSql: "SET SESSION ROLE postgres" }); + return run(session, file).pipe( + Effect.tap(() => + Effect.sync(() => { + const execs = calls.filter((call) => call.kind === "exec").map((call) => call.sql); + expect(execs.slice(-2)).toEqual(["DISCARD ALL", "SET SESSION ROLE postgres"]); + const batches = calls.filter((call) => call.kind === "batch"); + expect(batches[0]?.statements?.map(({ sql }) => sql)).toEqual(["select 1"]); + expect(batches[1]?.statements?.map(({ sql }) => sql)).toEqual([ + "select 2", + "SET SESSION ROLE postgres", + expect.stringContaining("supabase_migrations.schema_migrations"), + ]); + rmSync(dir, { recursive: true, force: true }); + }), + ), + ); + }); + + it.effect("sends no trailing restore when the file ends on a restored standalone", () => { + const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); + const file = join(dir, "20240101120000_discard_last.sql"); + writeFileSync(file, "select 1;\nDISCARD ALL;"); + const { session, calls } = fakeSession({ restoreRoleSql: "SET SESSION ROLE postgres" }); + return run(session, file).pipe( + Effect.tap(() => + Effect.sync(() => { + const execs = calls.filter((call) => call.kind === "exec").map((call) => call.sql); + expect(execs.slice(-2)).toEqual(["DISCARD ALL", "SET SESSION ROLE postgres"]); + const batches = calls.filter((call) => call.kind === "batch"); + expect(batches.at(-1)?.statements?.map(({ sql }) => sql)).toEqual([ + expect.stringContaining("supabase_migrations.schema_migrations"), + ]); + rmSync(dir, { recursive: true, force: true }); + }), + ), + ); + }); + it.effect("reports a mid-file restore's own failure at its host statement", () => { const dir = mkdtempSync(join(tmpdir(), "legacy-apply-")); const file = join(dir, "20240101120000_fail.sql"); @@ -1116,6 +1216,85 @@ describe("legacyIsPipelineIncompatible", () => { ["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], + ["create database", "CREATE DATABASE demo", true], + ["drop database", "DROP DATABASE IF EXISTS demo", true], + ["create tablespace", "CREATE TABLESPACE ts LOCATION '/tmp/ts'", true], + ["drop tablespace", "DROP TABLESPACE ts", true], + ["reindex database", "REINDEX DATABASE postgres", true], + ["reindex system with options", "REINDEX (VERBOSE) SYSTEM postgres", true], + ["reindex database adjacent options", "REINDEX(VERBOSE) DATABASE postgres", true], + ["reindex concurrently as option", "REINDEX (CONCURRENTLY) INDEX widgets_id_idx", true], + ["reindex mixed option list", "REINDEX (VERBOSE, CONCURRENTLY) TABLE public.widgets", true], + [ + "reindex concurrently false over-routes conservatively", + "REINDEX (CONCURRENTLY FALSE) INDEX widgets_id_idx", + true, + ], + ["reindex verbose option only", "REINDEX (VERBOSE) INDEX widgets_id_idx", false], + ["reindex schema", "REINDEX SCHEMA public", true], + ["reindex table non-concurrent", "REINDEX TABLE public.widgets", false], + ["alter database", "ALTER DATABASE demo SET search_path = public", false], + ["alter database set tablespace", "ALTER DATABASE demo SET TABLESPACE fast", true], + [ + "alter table all in tablespace", + "ALTER TABLE ALL IN TABLESPACE old_ts SET TABLESPACE fast", + true, + ], + [ + "alter index all in tablespace", + "ALTER INDEX ALL IN TABLESPACE old_ts SET TABLESPACE fast", + true, + ], + [ + "alter materialized view all in tablespace", + "ALTER MATERIALIZED VIEW ALL IN TABLESPACE old_ts SET TABLESPACE fast", + true, + ], + ["alter table set tablespace single", "ALTER TABLE t SET TABLESPACE fast", false], + ["alter database set tablespace multiline", "ALTER DATABASE demo\n SET TABLESPACE fast", true], + [ + "alter database with tablespace inside a literal over-routes conservatively", + "ALTER DATABASE demo SET application_name TO 'foo SET TABLESPACE bar'", + true, + ], + ["detach partition concurrently", "ALTER TABLE m DETACH PARTITION p CONCURRENTLY", true], + ["detach partition finalize stays batched", "ALTER TABLE m DETACH PARTITION p FINALIZE", false], + ["detach partition plain", "ALTER TABLE m DETACH PARTITION p", false], + [ + "detach inside a comment over-routes conservatively", + "ALTER TABLE m ADD COLUMN x int /* DETACH PARTITION p CONCURRENTLY */", + true, + ], + [ + "detach partition qualified concurrently", + "ALTER TABLE IF EXISTS ONLY s.m\n DETACH PARTITION p CONCURRENTLY", + true, + ], + [ + "detach partition quoted qualified concurrently", + 'ALTER TABLE "tenant schema".events DETACH PARTITION "p 1" CONCURRENTLY', + true, + ], + [ + "detach partition spaced qualification", + 'ALTER TABLE "tenant schema" . events DETACH PARTITION p CONCURRENTLY', + true, + ], + ["create subscription", "CREATE SUBSCRIPTION sub CONNECTION 'host=h' PUBLICATION pub", true], + ["drop subscription", "DROP SUBSCRIPTION IF EXISTS sub", true], + ["alter subscription", "ALTER SUBSCRIPTION sub DISABLE", false], + ["alter subscription refresh", "ALTER SUBSCRIPTION sub REFRESH PUBLICATION", true], + ["alter subscription set publication", "ALTER SUBSCRIPTION sub SET PUBLICATION p", true], + ["alter subscription refresh multiline", "ALTER SUBSCRIPTION sub\n REFRESH PUBLICATION", true], + ["alter subscription set options", "ALTER SUBSCRIPTION sub SET (slot_name = 's')", false], + ["discard all", "DISCARD ALL", true], + ["discard temp", "DISCARD TEMP", false], + [ + "refresh materialized view concurrently", + "REFRESH MATERIALIZED VIEW CONCURRENTLY public.mv", + true, + ], + ["plain refresh materialized view", "REFRESH MATERIALIZED VIEW public.mv", false], [ "lower-case create index concurrently", "create index concurrently widgets_id_idx on public.widgets(id)",