Prevent duplicate migration-stamp failures during concurrent startup (#1454)

* 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. * 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. ---------

Jad committed 2 months ago

Commit e8ea62c1 parent 69b0e643 files · +78 −6Browse files

Unified · Split

3 files

.changeset/concurrent-data-migration-stamps.md +5 −0
packages/core/sdk/src/sqlite-data-migrations.test.ts +61 −0
packages/core/sdk/src/sqlite-data-migrations.ts +12 −6

.changeset/concurrent-data-migration-stamps.md+5−0

@@ -0,0 +1,5 @@
	1	+---
	2	+"executor": patch
	3	+---
	4	+
	5	+Prevent concurrent SQLite data-migration runners from failing when another runner commits the same ledger stamp first.

packages/core/sdk/src/sqlite-data-migrations.test.ts+61−0

@@ -1,4 +1,5 @@
1	1	import { describe, expect, it } from "@effect/vitest";
	2	+import { createClient } from "@libsql/client";
2	3	import { Effect, Predicate } from "effect";
3	4
4	5	import {
@@ -76,6 +77,66 @@ describe("runSqliteDataMigrations", () => { · 71 unmodified lines
76	77	}),
77	78	);
78	79
	80	+ it.effect("converges when concurrent runners stamp the same migration", () =>
	81	+ Effect.gen(function* () {
	82	+ const client = createClient({ url: ":memory:" });
	83	+ let bodiesStarted = 0;
	84	+ let releaseBodies: (() => void) \| undefined;
	85	+ // Hold both bodies until both runners have read the ledger as empty.
	86	+ const bothBodiesStarted = new Promise<void>((resolve) => {
	87	+ releaseBodies = resolve;
	88	+ });
	89	+ const migration: SqliteDataMigration = {
	90	+ name: "2026-06-05-concurrent",
	91	+ run: () =>
	92	+ Effect.promise(() => {
	93	+ bodiesStarted++;
	94	+ if (bodiesStarted === 2) releaseBodies?.();
	95	+ return bothBodiesStarted;
	96	+ }),
	97	+ };
	98	+
	99	+ const results = yield* Effect.all(
	100	+ [
	101	+ runSqliteDataMigrations(client, [migration]),
	102	+ runSqliteDataMigrations(client, [migration]),
	103	+ ],
	104	+ { concurrency: "unbounded" },
	105	+ );
	106	+
	107	+ expect(results).toEqual([["2026-06-05-concurrent"], ["2026-06-05-concurrent"]]);
	108	+ expect(bodiesStarted).toBe(2);
	109	+ const stamps = yield* Effect.promise(() =>
	110	+ client.execute("SELECT name FROM data_migration ORDER BY name"),
	111	+ );
	112	+ expect(stamps.rows.map((row) => row.name)).toEqual(["2026-06-05-concurrent"]);
	113	+ client.close();
	114	+ }),
	115	+ );
	116	+
	117	+ it.effect("surfaces non-conflict stamp failures", () =>
	118	+ Effect.gen(function* () {
	119	+ const client: SqliteDataMigrationClient = {
	120	+ execute: (stmt) => {
	121	+ const sql = typeof stmt === "string" ? stmt : stmt.sql;
	122	+ if (typeof stmt === "object" && sql.startsWith("INSERT INTO data_migration")) {
	123	+ // oxlint-disable-next-line executor/no-promise-reject -- simulates a storage-driver rejection at the adapter boundary under test
	124	+ return Promise.reject("disk full");
	125	+ }
	126	+ return Promise.resolve({ rows: [] });
	127	+ },
	128	+ };
	129	+ const migration = migrationSpy("2026-06-05-unstamped");
	130	+
	131	+ const failure = yield* runSqliteDataMigrations(client, [migration.migration]).pipe(
	132	+ Effect.flip,
	133	+ );
	134	+
	135	+ expect(Predicate.isTagged(failure, "DataMigrationError")).toBe(true);
	136	+ expect((failure as DataMigrationError).cause).toBe("disk full");
	137	+ }),
	138	+ );
	139	+
79	140	it.effect("a failing migration leaves no stamp and surfaces the failure", () =>
80	141	Effect.gen(function* () {
81	142	const { client, stamps } = makeFakeClient([]);

packages/core/sdk/src/sqlite-data-migrations.ts+12−6

@@ -1,9 +1,9 @@
1	1	// ---------------------------------------------------------------------------
2		−// Stamped data-migration ledger for the libSQL-backed apps (local boot,
3		−// selfhost boot). Cloud runs schema + data migrations through its drizzle
4		−// chain out-of-band; the local apps have no operator, so their migrations
5		−// run at boot — and before this ledger existed, each one re-scanned its
6		−// tables on every startup to decide "did I already run?" by data shape.
	2	+// Stamped data-migration ledger for the SQLite-backed hosts (local,
	3	+// selfhost, Cloudflare D1). PostgreSQL cloud runs schema + data migrations
	4	+// through its drizzle chain out-of-band; the SQLite hosts run them at boot —
	5	+// and before this ledger existed, each one re-scanned its tables on every
	6	+// startup to decide "did I already run?" by data shape.
7	7	// That accumulates (N migrations = N full-table scans per boot, forever)
8	8	// and makes idempotence a per-migration proof obligation.
9	9	//
@@ -112,10 +112,16 @@ export const runSqliteDataMigrations = ( · 102 unmodified lines
112	112	for (const migration of migrations) {
113	113	if (completed.has(migration.name)) continue;
114	114	yield* migration.run(client);
	115	+ // Multiple hosts can boot against the same database and observe the same
	116	+ // migration as pending. Their idempotent bodies may both finish, but only
	117	+ // one ledger insert can win. Ignore only a conflict on this exact stamp;
	118	+ // every other ledger failure still fails the boot.
115	119	yield* execute(
116	120	client,
117	121	{
118		− sql: `INSERT INTO ${LEDGER_TABLE} (name, time_completed) VALUES (?, ?)`,
	122	+ sql: `INSERT INTO ${LEDGER_TABLE} (name, time_completed)
	123	+ VALUES (?, ?)
	124	+ ON CONFLICT(name) DO NOTHING`,
119	125	args: [migration.name, Date.now()],
120	126	},
121	127	migration.name,