Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/concurrent-data-migration-stamps.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"executor": patch
---

Prevent concurrent SQLite data-migration runners from failing when another runner commits the same ledger stamp first.
61 changes: 61 additions & 0 deletions packages/core/sdk/src/sqlite-data-migrations.test.ts
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import { describe, expect, it } from "@effect/vitest";
import { createClient } from "@libsql/client";
import { Effect, Predicate } from "effect";

import {
Expand Down Expand Up @@ -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<void>((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([]);
Expand Down
18 changes: 12 additions & 6 deletions packages/core/sdk/src/sqlite-data-migrations.ts
Original file line number Diff line number Diff line change
@@ -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.
//
Expand Down Expand Up @@ -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,
Expand Down
Loading