diff --git a/.changeset/concurrent-data-migration-stamps.md b/.changeset/concurrent-data-migration-stamps.md new file mode 100644 index 000000000..f040a62a7 --- /dev/null +++ b/.changeset/concurrent-data-migration-stamps.md @@ -0,0 +1,5 @@ +--- +"executor": patch +--- + +Prevent concurrent SQLite data-migration runners from failing when another runner commits the same ledger stamp first. diff --git a/packages/core/sdk/src/sqlite-data-migrations.test.ts b/packages/core/sdk/src/sqlite-data-migrations.test.ts index ab40ffecd..b0d42c3b0 100644 --- a/packages/core/sdk/src/sqlite-data-migrations.test.ts +++ b/packages/core/sdk/src/sqlite-data-migrations.test.ts @@ -1,4 +1,5 @@ import { describe, expect, it } from "@effect/vitest"; +import { createClient } from "@libsql/client"; import { Effect, Predicate } from "effect"; import { @@ -76,6 +77,66 @@ describe("runSqliteDataMigrations", () => { }), ); + it.effect("converges when concurrent runners stamp the same migration", () => + Effect.gen(function* () { + const client = createClient({ url: ":memory:" }); + let bodiesStarted = 0; + let releaseBodies: (() => void) | undefined; + // Hold both bodies until both runners have read the ledger as empty. + const bothBodiesStarted = new Promise((resolve) => { + releaseBodies = resolve; + }); + const migration: SqliteDataMigration = { + name: "2026-06-05-concurrent", + run: () => + Effect.promise(() => { + bodiesStarted++; + if (bodiesStarted === 2) releaseBodies?.(); + return bothBodiesStarted; + }), + }; + + const results = yield* Effect.all( + [ + runSqliteDataMigrations(client, [migration]), + runSqliteDataMigrations(client, [migration]), + ], + { concurrency: "unbounded" }, + ); + + expect(results).toEqual([["2026-06-05-concurrent"], ["2026-06-05-concurrent"]]); + expect(bodiesStarted).toBe(2); + const stamps = yield* Effect.promise(() => + client.execute("SELECT name FROM data_migration ORDER BY name"), + ); + expect(stamps.rows.map((row) => row.name)).toEqual(["2026-06-05-concurrent"]); + client.close(); + }), + ); + + it.effect("surfaces non-conflict stamp failures", () => + Effect.gen(function* () { + const client: SqliteDataMigrationClient = { + execute: (stmt) => { + const sql = typeof stmt === "string" ? stmt : stmt.sql; + if (typeof stmt === "object" && sql.startsWith("INSERT INTO data_migration")) { + // oxlint-disable-next-line executor/no-promise-reject -- simulates a storage-driver rejection at the adapter boundary under test + return Promise.reject("disk full"); + } + return Promise.resolve({ rows: [] }); + }, + }; + const migration = migrationSpy("2026-06-05-unstamped"); + + const failure = yield* runSqliteDataMigrations(client, [migration.migration]).pipe( + Effect.flip, + ); + + expect(Predicate.isTagged(failure, "DataMigrationError")).toBe(true); + expect((failure as DataMigrationError).cause).toBe("disk full"); + }), + ); + it.effect("a failing migration leaves no stamp and surfaces the failure", () => Effect.gen(function* () { const { client, stamps } = makeFakeClient([]); diff --git a/packages/core/sdk/src/sqlite-data-migrations.ts b/packages/core/sdk/src/sqlite-data-migrations.ts index 4ebdcd178..60954d3cf 100644 --- a/packages/core/sdk/src/sqlite-data-migrations.ts +++ b/packages/core/sdk/src/sqlite-data-migrations.ts @@ -1,9 +1,9 @@ // --------------------------------------------------------------------------- -// Stamped data-migration ledger for the libSQL-backed apps (local boot, -// selfhost boot). Cloud runs schema + data migrations through its drizzle -// chain out-of-band; the local apps have no operator, so their migrations -// run at boot — and before this ledger existed, each one re-scanned its -// tables on every startup to decide "did I already run?" by data shape. +// Stamped data-migration ledger for the SQLite-backed hosts (local, +// selfhost, Cloudflare D1). PostgreSQL cloud runs schema + data migrations +// through its drizzle chain out-of-band; the SQLite hosts run them at boot — +// and before this ledger existed, each one re-scanned its tables on every +// startup to decide "did I already run?" by data shape. // That accumulates (N migrations = N full-table scans per boot, forever) // and makes idempotence a per-migration proof obligation. // @@ -112,10 +112,16 @@ export const runSqliteDataMigrations = ( for (const migration of migrations) { if (completed.has(migration.name)) continue; yield* migration.run(client); + // Multiple hosts can boot against the same database and observe the same + // migration as pending. Their idempotent bodies may both finish, but only + // one ledger insert can win. Ignore only a conflict on this exact stamp; + // every other ledger failure still fails the boot. yield* execute( client, { - sql: `INSERT INTO ${LEDGER_TABLE} (name, time_completed) VALUES (?, ?)`, + sql: `INSERT INTO ${LEDGER_TABLE} (name, time_completed) + VALUES (?, ?) + ON CONFLICT(name) DO NOTHING`, args: [migration.name, Date.now()], }, migration.name,