Skip to content

Commit 4108abb

Browse files
jadchjadch
committed
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.
1 parent f732009 commit 4108abb

3 files changed

Lines changed: 36 additions & 65 deletions

File tree

‎.changeset/concurrent-data-migration-stamps.md‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
---
2-
"@executor-js/sdk": patch
2+
"executor": patch
33
---
44

5-
Allow concurrent data-migration runners to converge when another runner commits the same ledger stamp first.
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‎

Lines changed: 24 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
import { describe, expect, it } from "@effect/vitest";
2+
import { createClient } from "@libsql/client";
23
import { Effect, Predicate } from "effect";
34

45
import {
@@ -78,57 +79,42 @@ describe("runSqliteDataMigrations", () => {
7879

7980
it.effect("converges when concurrent runners stamp the same migration", () =>
8081
Effect.gen(function* () {
81-
const stamps = new Set<string>();
82-
let initialReads = 0;
83-
let releaseInitialReads: (() => void) | undefined;
84-
const initialReadsComplete = new Promise<void>((resolve) => {
85-
releaseInitialReads = resolve;
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;
8688
});
87-
const client: SqliteDataMigrationClient = {
88-
execute: (stmt) => {
89-
const sql = typeof stmt === "string" ? stmt : stmt.sql;
90-
if (sql === "SELECT name FROM data_migration") {
91-
const rows = [...stamps].map((name) => ({ name }));
92-
initialReads++;
93-
if (initialReads === 2) releaseInitialReads?.();
94-
return initialReadsComplete.then(() => ({ rows }));
95-
}
96-
if (typeof stmt === "object" && sql.startsWith("INSERT INTO data_migration")) {
97-
const name = String(stmt.args[0]);
98-
if (stamps.has(name)) {
99-
// oxlint-disable-next-line executor/no-promise-reject -- simulates a storage-driver rejection at the adapter boundary under test
100-
return Promise.reject("UNIQUE constraint failed: data_migration.name");
101-
}
102-
stamps.add(name);
103-
return Promise.resolve({ rows: [] });
104-
}
105-
if (
106-
typeof stmt === "object" &&
107-
sql === "SELECT name FROM data_migration WHERE name = ?"
108-
) {
109-
const name = String(stmt.args[0]);
110-
return Promise.resolve({ rows: stamps.has(name) ? [{ name }] : [] });
111-
}
112-
return Promise.resolve({ rows: [] });
113-
},
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+
}),
11497
};
115-
const migration = migrationSpy("2026-06-05-concurrent");
11698

11799
const results = yield* Effect.all(
118100
[
119-
runSqliteDataMigrations(client, [migration.migration]),
120-
runSqliteDataMigrations(client, [migration.migration]),
101+
runSqliteDataMigrations(client, [migration]),
102+
runSqliteDataMigrations(client, [migration]),
121103
],
122104
{ concurrency: "unbounded" },
123105
);
124106

125107
expect(results).toEqual([["2026-06-05-concurrent"], ["2026-06-05-concurrent"]]);
126-
expect(migration.calls.length).toBe(2);
127-
expect(stamps).toEqual(new Set(["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();
128114
}),
129115
);
130116

131-
it.effect("surfaces a stamp failure when no concurrent runner committed it", () =>
117+
it.effect("surfaces non-conflict stamp failures", () =>
132118
Effect.gen(function* () {
133119
const client: SqliteDataMigrationClient = {
134120
execute: (stmt) => {

‎packages/core/sdk/src/sqlite-data-migrations.ts‎

Lines changed: 10 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,9 +1,9 @@
11
// ---------------------------------------------------------------------------
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.
77
// That accumulates (N migrations = N full-table scans per boot, forever)
88
// and makes idempotence a per-migration proof obligation.
99
//
@@ -114,32 +114,17 @@ export const runSqliteDataMigrations = (
114114
yield* migration.run(client);
115115
// Multiple hosts can boot against the same database and observe the same
116116
// migration as pending. Their idempotent bodies may both finish, but only
117-
// one ledger insert can win. Treat the losing insert as success only after
118-
// verifying that another runner committed this exact stamp.
117+
// one ledger insert can win. Ignore only a conflict on this exact stamp;
118+
// every other ledger failure still fails the boot.
119119
yield* execute(
120120
client,
121121
{
122-
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`,
123125
args: [migration.name, Date.now()],
124126
},
125127
migration.name,
126-
).pipe(
127-
Effect.catch((stampError) =>
128-
execute(
129-
client,
130-
{
131-
sql: `SELECT name FROM ${LEDGER_TABLE} WHERE name = ?`,
132-
args: [migration.name],
133-
},
134-
migration.name,
135-
).pipe(
136-
Effect.flatMap((result) =>
137-
result.rows.some((row) => row.name === migration.name)
138-
? Effect.void
139-
: Effect.fail(stampError),
140-
),
141-
),
142-
),
143128
);
144129
applied.push(migration.name);
145130
}

0 commit comments

Comments
 (0)