From f732009d0691405419db5cdedc47a3adb3ac1290 Mon Sep 17 00:00:00 2001 From: Jad Date: Thu, 23 Jul 2026 08:16:01 +0200 Subject: [PATCH 1/2] Converge concurrent data migration stamps When multiple hosts boot against the same database, let a runner accept a failed ledger insert only after confirming another runner committed the exact migration stamp. Preserve unrelated stamp failures and cover both outcomes with concurrent regressions. --- .../concurrent-data-migration-stamps.md | 5 ++ .../sdk/src/sqlite-data-migrations.test.ts | 75 +++++++++++++++++++ .../core/sdk/src/sqlite-data-migrations.ts | 21 ++++++ 3 files changed, 101 insertions(+) create mode 100644 .changeset/concurrent-data-migration-stamps.md diff --git a/.changeset/concurrent-data-migration-stamps.md b/.changeset/concurrent-data-migration-stamps.md new file mode 100644 index 000000000..e631199a5 --- /dev/null +++ b/.changeset/concurrent-data-migration-stamps.md @@ -0,0 +1,5 @@ +--- +"@executor-js/sdk": patch +--- + +Allow concurrent data-migration runners to converge 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..b6f00d165 100644 --- a/packages/core/sdk/src/sqlite-data-migrations.test.ts +++ b/packages/core/sdk/src/sqlite-data-migrations.test.ts @@ -76,6 +76,81 @@ describe("runSqliteDataMigrations", () => { }), ); + it.effect("converges when concurrent runners stamp the same migration", () => + Effect.gen(function* () { + const stamps = new Set(); + let initialReads = 0; + let releaseInitialReads: (() => void) | undefined; + const initialReadsComplete = new Promise((resolve) => { + releaseInitialReads = resolve; + }); + const client: SqliteDataMigrationClient = { + execute: (stmt) => { + const sql = typeof stmt === "string" ? stmt : stmt.sql; + if (sql === "SELECT name FROM data_migration") { + const rows = [...stamps].map((name) => ({ name })); + initialReads++; + if (initialReads === 2) releaseInitialReads?.(); + return initialReadsComplete.then(() => ({ rows })); + } + if (typeof stmt === "object" && sql.startsWith("INSERT INTO data_migration")) { + const name = String(stmt.args[0]); + if (stamps.has(name)) { + // oxlint-disable-next-line executor/no-promise-reject -- simulates a storage-driver rejection at the adapter boundary under test + return Promise.reject("UNIQUE constraint failed: data_migration.name"); + } + stamps.add(name); + return Promise.resolve({ rows: [] }); + } + if ( + typeof stmt === "object" && + sql === "SELECT name FROM data_migration WHERE name = ?" + ) { + const name = String(stmt.args[0]); + return Promise.resolve({ rows: stamps.has(name) ? [{ name }] : [] }); + } + return Promise.resolve({ rows: [] }); + }, + }; + const migration = migrationSpy("2026-06-05-concurrent"); + + const results = yield* Effect.all( + [ + runSqliteDataMigrations(client, [migration.migration]), + runSqliteDataMigrations(client, [migration.migration]), + ], + { concurrency: "unbounded" }, + ); + + expect(results).toEqual([["2026-06-05-concurrent"], ["2026-06-05-concurrent"]]); + expect(migration.calls.length).toBe(2); + expect(stamps).toEqual(new Set(["2026-06-05-concurrent"])); + }), + ); + + it.effect("surfaces a stamp failure when no concurrent runner committed it", () => + 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..779e73326 100644 --- a/packages/core/sdk/src/sqlite-data-migrations.ts +++ b/packages/core/sdk/src/sqlite-data-migrations.ts @@ -112,6 +112,10 @@ 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. Treat the losing insert as success only after + // verifying that another runner committed this exact stamp. yield* execute( client, { @@ -119,6 +123,23 @@ export const runSqliteDataMigrations = ( args: [migration.name, Date.now()], }, migration.name, + ).pipe( + Effect.catch((stampError) => + execute( + client, + { + sql: `SELECT name FROM ${LEDGER_TABLE} WHERE name = ?`, + args: [migration.name], + }, + migration.name, + ).pipe( + Effect.flatMap((result) => + result.rows.some((row) => row.name === migration.name) + ? Effect.void + : Effect.fail(stampError), + ), + ), + ), ); applied.push(migration.name); } From 4108abb04721853f8d133f26392a7d45fef41f89 Mon Sep 17 00:00:00 2001 From: Jad Date: Thu, 23 Jul 2026 09:00:00 +0200 Subject: [PATCH 2/2] Use atomic conflict handling for migration stamps Resolve duplicate stamp races in one SQLite statement, avoiding a cross-session verification read. Exercise the behavior against real libSQL and align the changeset with repository conventions. --- .../concurrent-data-migration-stamps.md | 4 +- .../sdk/src/sqlite-data-migrations.test.ts | 62 +++++++------------ .../core/sdk/src/sqlite-data-migrations.ts | 35 +++-------- 3 files changed, 36 insertions(+), 65 deletions(-) diff --git a/.changeset/concurrent-data-migration-stamps.md b/.changeset/concurrent-data-migration-stamps.md index e631199a5..f040a62a7 100644 --- a/.changeset/concurrent-data-migration-stamps.md +++ b/.changeset/concurrent-data-migration-stamps.md @@ -1,5 +1,5 @@ --- -"@executor-js/sdk": patch +"executor": patch --- -Allow concurrent data-migration runners to converge when another runner commits the same ledger stamp first. +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 b6f00d165..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 { @@ -78,57 +79,42 @@ describe("runSqliteDataMigrations", () => { it.effect("converges when concurrent runners stamp the same migration", () => Effect.gen(function* () { - const stamps = new Set(); - let initialReads = 0; - let releaseInitialReads: (() => void) | undefined; - const initialReadsComplete = new Promise((resolve) => { - releaseInitialReads = resolve; + 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 client: SqliteDataMigrationClient = { - execute: (stmt) => { - const sql = typeof stmt === "string" ? stmt : stmt.sql; - if (sql === "SELECT name FROM data_migration") { - const rows = [...stamps].map((name) => ({ name })); - initialReads++; - if (initialReads === 2) releaseInitialReads?.(); - return initialReadsComplete.then(() => ({ rows })); - } - if (typeof stmt === "object" && sql.startsWith("INSERT INTO data_migration")) { - const name = String(stmt.args[0]); - if (stamps.has(name)) { - // oxlint-disable-next-line executor/no-promise-reject -- simulates a storage-driver rejection at the adapter boundary under test - return Promise.reject("UNIQUE constraint failed: data_migration.name"); - } - stamps.add(name); - return Promise.resolve({ rows: [] }); - } - if ( - typeof stmt === "object" && - sql === "SELECT name FROM data_migration WHERE name = ?" - ) { - const name = String(stmt.args[0]); - return Promise.resolve({ rows: stamps.has(name) ? [{ name }] : [] }); - } - return Promise.resolve({ rows: [] }); - }, + const migration: SqliteDataMigration = { + name: "2026-06-05-concurrent", + run: () => + Effect.promise(() => { + bodiesStarted++; + if (bodiesStarted === 2) releaseBodies?.(); + return bothBodiesStarted; + }), }; - const migration = migrationSpy("2026-06-05-concurrent"); const results = yield* Effect.all( [ - runSqliteDataMigrations(client, [migration.migration]), - runSqliteDataMigrations(client, [migration.migration]), + runSqliteDataMigrations(client, [migration]), + runSqliteDataMigrations(client, [migration]), ], { concurrency: "unbounded" }, ); expect(results).toEqual([["2026-06-05-concurrent"], ["2026-06-05-concurrent"]]); - expect(migration.calls.length).toBe(2); - expect(stamps).toEqual(new Set(["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 a stamp failure when no concurrent runner committed it", () => + it.effect("surfaces non-conflict stamp failures", () => Effect.gen(function* () { const client: SqliteDataMigrationClient = { execute: (stmt) => { diff --git a/packages/core/sdk/src/sqlite-data-migrations.ts b/packages/core/sdk/src/sqlite-data-migrations.ts index 779e73326..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. // @@ -114,32 +114,17 @@ export const runSqliteDataMigrations = ( 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. Treat the losing insert as success only after - // verifying that another runner committed this exact stamp. + // 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, - ).pipe( - Effect.catch((stampError) => - execute( - client, - { - sql: `SELECT name FROM ${LEDGER_TABLE} WHERE name = ?`, - args: [migration.name], - }, - migration.name, - ).pipe( - Effect.flatMap((result) => - result.rows.some((row) => row.name === migration.name) - ? Effect.void - : Effect.fail(stampError), - ), - ), - ), ); applied.push(migration.name); }