From 85790068a0286002a46c3d25366632283a5270a1 Mon Sep 17 00:00:00 2001 From: "Yifeng \"Evan\" Wang" <7312949+doodlewind@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:11:11 +0800 Subject: [PATCH 1/2] fix: apply remote D1 migrations atomically --- .github/workflows/deploy-web.yml | 8 +- apps/backend/package.json | 4 +- apps/backend/scripts/d1-migrate-remote.mjs | 323 +++++++++++++++++++ apps/backend/tests/d1-migrate-remote.test.ts | 226 +++++++++++++ 4 files changed, 556 insertions(+), 5 deletions(-) create mode 100644 apps/backend/scripts/d1-migrate-remote.mjs create mode 100644 apps/backend/tests/d1-migrate-remote.test.ts diff --git a/.github/workflows/deploy-web.yml b/.github/workflows/deploy-web.yml index 01e1d149..b20693a6 100644 --- a/.github/workflows/deploy-web.yml +++ b/.github/workflows/deploy-web.yml @@ -196,9 +196,11 @@ jobs: - name: Build web router run: pnpm --filter @spool/web run build - # Apply every committed D1 migration before publishing code that may - # depend on the new schema. A failed migration stops the job, leaving - # the currently deployed backend and router untouched. + # Apply every committed D1 migration through D1's atomic import endpoint + # before publishing code that may depend on the new schema. The standard + # /query migration path cannot reliably ingest trigger-heavy migrations. + # A failed import rolls the database back and stops the deployment, + # leaving the currently deployed backend and router untouched. - name: Apply staging D1 migrations if: env.DEPLOY_TARGET == 'staging' env: diff --git a/apps/backend/package.json b/apps/backend/package.json index 48535092..c848f3b6 100644 --- a/apps/backend/package.json +++ b/apps/backend/package.json @@ -22,8 +22,8 @@ "deploy:prod": "node scripts/deploy-pages.mjs production", "deploy:staging": "node scripts/deploy-pages.mjs staging", "d1:migrate:local": "wrangler d1 migrations apply spool-share-db --local --config wrangler.toml", - "d1:migrate:prod": "wrangler d1 migrations apply spool-share-db --remote --config wrangler.prod.toml", - "d1:migrate:staging": "wrangler d1 migrations apply spool-share-db-staging --remote --config wrangler.staging.toml", + "d1:migrate:prod": "node scripts/d1-migrate-remote.mjs --database spool-share-db --config wrangler.prod.toml", + "d1:migrate:staging": "node scripts/d1-migrate-remote.mjs --database spool-share-db-staging --config wrangler.staging.toml", "schema:smoke": "node scripts/schema-smoke.mjs", "test": "vp test run", "typecheck": "tsc --noEmit" diff --git a/apps/backend/scripts/d1-migrate-remote.mjs b/apps/backend/scripts/d1-migrate-remote.mjs new file mode 100644 index 00000000..a2c6e6a7 --- /dev/null +++ b/apps/backend/scripts/d1-migrate-remote.mjs @@ -0,0 +1,323 @@ +import { spawn } from 'node:child_process' +import { mkdtemp, readFile, readdir, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { dirname, isAbsolute, join, resolve } from 'node:path' +import { fileURLToPath } from 'node:url' + +const appDir = dirname(dirname(fileURLToPath(import.meta.url))) +const wranglerEntry = join(appDir, 'node_modules', 'wrangler', 'bin', 'wrangler.js') + +export const MIGRATIONS_TABLE_SQL = `CREATE TABLE IF NOT EXISTS "d1_migrations"( + id INTEGER PRIMARY KEY AUTOINCREMENT, + name TEXT UNIQUE, + applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP NOT NULL +)` + +const MIGRATION_NAME = /^\d{4}_[A-Za-z0-9_.-]+\.sql$/ +const GUARD_TABLE = '__spool_d1_migration_guard' + +export function parseArgs(argv) { + const options = { + config: undefined, + database: undefined, + migrationsDir: join(appDir, 'migrations'), + } + + for (let index = 0; index < argv.length; index += 1) { + const value = argv[index] + const next = argv[index + 1] + if (value === '--database' && next) { + options.database = next + index += 1 + continue + } + if (value === '--config' && next) { + options.config = next + index += 1 + continue + } + if (value === '--migrations-dir' && next) { + options.migrationsDir = isAbsolute(next) ? next : resolve(appDir, next) + index += 1 + continue + } + throw new Error(`Unknown or incomplete argument: ${value}`) + } + + if (!options.database || !options.config) { + throw new Error( + 'Usage: d1-migrate-remote.mjs --database --config [--migrations-dir ]', + ) + } + + return options +} + +export async function listMigrationNames(migrationsDir) { + const entries = await readdir(migrationsDir, { withFileTypes: true }) + const names = entries + .filter((entry) => entry.isFile() && entry.name.endsWith('.sql')) + .map((entry) => entry.name) + .sort() + + if (names.length === 0) { + throw new Error(`No SQL migrations found in ${migrationsDir}`) + } + + const invalid = names.find((name) => !MIGRATION_NAME.test(name)) + if (invalid) { + throw new Error(`Migration filename is not safe for the remote ledger: ${invalid}`) + } + + return names +} + +export function validateLedgerPrefix(migrationNames, appliedRows) { + if (!Array.isArray(appliedRows)) { + throw new TypeError('D1 returned an invalid migration ledger') + } + if (appliedRows.length > migrationNames.length) { + throw new Error( + `Remote migration ledger has ${appliedRows.length} entries but only ${migrationNames.length} local migrations exist`, + ) + } + + for (let index = 0; index < appliedRows.length; index += 1) { + const actual = appliedRows[index]?.name + const expected = migrationNames[index] + if (actual !== expected) { + throw new Error( + `Remote migration ledger diverges at position ${index + 1}: expected ${expected}, received ${String(actual)}`, + ) + } + } + + return appliedRows.length +} + +function sqlString(value) { + return `'${value.replaceAll("'", "''")}'` +} + +export function buildMigrationImport(contents, migrationName, expectedCount = 0) { + if (!MIGRATION_NAME.test(migrationName)) { + throw new Error(`Migration filename is not safe for the remote ledger: ${migrationName}`) + } + if (!Number.isSafeInteger(expectedCount) || expectedCount < 0) { + throw new Error(`Invalid expected migration count: ${String(expectedCount)}`) + } + + const source = Buffer.isBuffer(contents) ? contents : Buffer.from(contents) + const sql = source.toString('utf8') + if (source.length === 0 || sql.trim().length === 0) { + throw new Error(`Migration is empty: ${migrationName}`) + } + if (/\bBEGIN\s+TRANSACTION\b/i.test(sql) || /\bCOMMIT\s*;/i.test(sql)) { + throw new Error( + `Migration ${migrationName} contains an explicit transaction; D1 imports are already atomic`, + ) + } + if (!sql.trimEnd().endsWith(';')) { + throw new Error(`Migration must end with a SQL statement terminator: ${migrationName}`) + } + + const separator = source.length > 0 && source[source.length - 1] === 0x0a ? '' : '\n' + const ledger = Buffer.from( + `${separator}\n-- Fail atomically if another runner advanced the ledger after our prefix read.\n` + + `CREATE TABLE "${GUARD_TABLE}" (\n` + + ` expected_count INTEGER NOT NULL,\n` + + ` actual_count INTEGER NOT NULL,\n` + + ` CHECK (expected_count = actual_count)\n` + + `);\n` + + `INSERT INTO "${GUARD_TABLE}" (expected_count, actual_count)\n` + + `SELECT ${expectedCount}, COUNT(*) FROM "d1_migrations";\n` + + `DROP TABLE "${GUARD_TABLE}";\n\n` + + `-- Recorded atomically with the schema and data changes above.\n` + + `INSERT INTO "d1_migrations" (name) VALUES (${sqlString(migrationName)});\n`, + ) + return Buffer.concat([source, ledger]) +} + +function formatProcessFailure(args, code, stdout, stderr) { + const output = [stdout.trim(), stderr.trim()].filter(Boolean).join('\n').slice(0, 8_000) + return new Error( + `wrangler ${args.join(' ')} failed with exit code ${String(code)}${output ? `\n${output}` : ''}`, + ) +} + +export function runWrangler(args) { + return new Promise((resolvePromise, reject) => { + const child = spawn(process.execPath, [wranglerEntry, ...args], { + cwd: appDir, + env: { + ...process.env, + NO_COLOR: '1', + WRANGLER_SEND_METRICS: 'false', + }, + stdio: ['ignore', 'pipe', 'pipe'], + }) + + let stdout = '' + let stderr = '' + child.stdout.setEncoding('utf8') + child.stderr.setEncoding('utf8') + child.stdout.on('data', (chunk) => { + stdout += chunk + }) + child.stderr.on('data', (chunk) => { + stderr += chunk + }) + + child.once('error', reject) + child.once('close', (code) => { + if (code === 0) { + resolvePromise({ stderr, stdout }) + return + } + reject(formatProcessFailure(args, code, stdout, stderr)) + }) + }) +} + +export function parseWranglerJson(stdout) { + let payload + try { + payload = JSON.parse(stdout) + } catch (error) { + throw new Error('Wrangler returned invalid JSON', { cause: error }) + } + if ( + !Array.isArray(payload) || + payload.length !== 1 || + payload.some((result) => result?.success !== true || !Array.isArray(result.results)) + ) { + throw new Error('Wrangler returned an unsuccessful D1 response') + } + return payload +} + +async function executeCommand({ config, database }, command) { + const { stdout } = await runWrangler([ + 'd1', + 'execute', + database, + '--remote', + '--config', + config, + '--command', + command, + '--json', + ]) + return parseWranglerJson(stdout) +} + +async function readLedger(options) { + const payload = await executeCommand(options, 'SELECT id, name FROM "d1_migrations" ORDER BY id') + const rows = payload.flatMap((execution) => execution.results) + if ( + rows.some( + (row) => + !row || + !Number.isSafeInteger(row.id) || + row.id <= 0 || + typeof row.name !== 'string' || + row.name.length === 0, + ) + ) { + throw new Error('Wrangler returned an invalid D1 migration ledger') + } + return rows +} + +async function executeImport({ config, database }, file) { + await runWrangler([ + 'd1', + 'execute', + database, + '--remote', + '--config', + config, + '--file', + file, + '--yes', + ]) +} + +export async function migrateRemote( + options, + dependencies = { executeCommand, executeImport, readLedger }, +) { + const migrationNames = await listMigrationNames(options.migrationsDir) + + await dependencies.executeCommand(options, MIGRATIONS_TABLE_SQL) + let appliedRows = await dependencies.readLedger(options) + let appliedCount = validateLedgerPrefix(migrationNames, appliedRows) + + if (appliedCount === migrationNames.length) { + console.log(`D1 migration ledger is current (${appliedCount}/${migrationNames.length}).`) + return + } + + for (let index = appliedCount; index < migrationNames.length; index += 1) { + const migrationName = migrationNames[index] + + // Another queued deployment may have completed while this job was waiting. + appliedRows = await dependencies.readLedger(options) + appliedCount = validateLedgerPrefix(migrationNames, appliedRows) + if (appliedCount > index) { + continue + } + if (appliedCount !== index) { + throw new Error( + `Remote migration ledger advanced unexpectedly: expected ${index} applied migrations, received ${appliedCount}`, + ) + } + + const directory = await mkdtemp(join(tmpdir(), 'spool-d1-migration-')) + const importFile = join(directory, migrationName) + try { + const contents = await readFile(join(options.migrationsDir, migrationName)) + await writeFile(importFile, buildMigrationImport(contents, migrationName, index), { + mode: 0o600, + }) + console.log(`Applying ${migrationName} through D1's atomic import endpoint...`) + try { + await dependencies.executeImport(options, importFile) + } catch (error) { + // A concurrent deployment can win the UNIQUE ledger insert. Accept + // that failure only when the remote ledger proves the same migration + // committed atomically; every other failure remains fatal. + const rowsAfterFailure = await dependencies.readLedger(options) + const countAfterFailure = validateLedgerPrefix(migrationNames, rowsAfterFailure) + if (countAfterFailure <= index) { + throw error + } + console.log(`${migrationName} was committed by another deployment.`) + } + } finally { + await rm(directory, { force: true, recursive: true }) + } + } + + const finalRows = await dependencies.readLedger(options) + const finalCount = validateLedgerPrefix(migrationNames, finalRows) + if (finalCount !== migrationNames.length) { + throw new Error( + `Remote migration verification failed: expected ${migrationNames.length}, received ${finalCount}`, + ) + } + console.log(`D1 migrations applied and verified (${finalCount}/${migrationNames.length}).`) +} + +async function main() { + const options = parseArgs(process.argv.slice(2)) + await migrateRemote(options) +} + +const invokedPath = process.argv[1] ? resolve(process.argv[1]) : '' +if (invokedPath === fileURLToPath(import.meta.url)) { + main().catch((error) => { + console.error(error instanceof Error ? error.message : String(error)) + process.exitCode = 1 + }) +} diff --git a/apps/backend/tests/d1-migrate-remote.test.ts b/apps/backend/tests/d1-migrate-remote.test.ts new file mode 100644 index 00000000..d07f941a --- /dev/null +++ b/apps/backend/tests/d1-migrate-remote.test.ts @@ -0,0 +1,226 @@ +import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' +import { tmpdir } from 'node:os' +import { join } from 'node:path' + +import { describe, expect, it, vi } from 'vite-plus/test' + +// The deployment utility is plain ESM so CI can execute it without a build step. +// @ts-expect-error The checked-in operational script deliberately has no TS build artifact. +import * as remoteMigrations from '../scripts/d1-migrate-remote.mjs' + +const { + MIGRATIONS_TABLE_SQL, + buildMigrationImport, + listMigrationNames, + migrateRemote, + parseArgs, + parseWranglerJson, + validateLedgerPrefix, +} = remoteMigrations + +describe('remote D1 migration runner', () => { + it('requires an explicit database and Wrangler config', () => { + expect(() => parseArgs([])).toThrow('Usage:') + expect( + parseArgs(['--database', 'example-db', '--config', 'wrangler.example.toml']), + ).toMatchObject({ + config: 'wrangler.example.toml', + database: 'example-db', + }) + expect(() => parseArgs(['--wat'])).toThrow('Unknown or incomplete argument') + }) + + it('keeps production and staging on the atomic remote runner', async () => { + const packageJson = JSON.parse( + await readFile(new URL('../package.json', import.meta.url), 'utf8'), + ) as { + scripts: Record + } + + for (const name of ['d1:migrate:prod', 'd1:migrate:staging']) { + expect(packageJson.scripts[name]).toContain('scripts/d1-migrate-remote.mjs') + expect(packageJson.scripts[name]).not.toContain('migrations apply') + } + }) + + it('fails closed on malformed or incomplete Wrangler JSON', () => { + expect(() => parseWranglerJson('not json')).toThrow('invalid JSON') + expect(() => parseWranglerJson('[]')).toThrow('unsuccessful D1 response') + expect(() => parseWranglerJson('[{"success":true}]')).toThrow('unsuccessful D1 response') + expect(parseWranglerJson('[{"success":true,"results":[]}]')).toEqual([ + { results: [], success: true }, + ]) + }) + + it('preserves migration bytes and appends the ledger write', () => { + const source = Buffer.from([ + 0xef, 0xbb, 0xbf, 0x53, 0x45, 0x4c, 0x45, 0x43, 0x54, 0x20, 0x31, 0x3b, + ]) + const result = buildMigrationImport(source, '0014_projects.sql', 13) + + expect(result.subarray(0, source.length)).toEqual(source) + expect(result.toString('utf8')).toContain('SELECT 13, COUNT(*) FROM "d1_migrations";') + expect(result.toString('utf8')).toContain( + `INSERT INTO "d1_migrations" (name) VALUES ('0014_projects.sql');`, + ) + }) + + it('rejects explicit transactions because D1 imports own the transaction', () => { + expect(() => + buildMigrationImport('BEGIN TRANSACTION; SELECT 1; COMMIT;', '0001_bad.sql'), + ).toThrow('already atomic') + }) + + it('rejects an unterminated migration before touching D1', () => { + expect(() => buildMigrationImport('SELECT 1', '0001_bad.sql')).toThrow( + 'must end with a SQL statement terminator', + ) + }) + + it('accepts only an exact prefix of the local migration order', () => { + const migrations = ['0001_init.sql', '0002_projects.sql'] + + expect(validateLedgerPrefix(migrations, [{ name: '0001_init.sql' }])).toBe(1) + expect(() => validateLedgerPrefix(migrations, [{ name: '0002_projects.sql' }])).toThrow( + 'diverges at position 1', + ) + expect(() => + validateLedgerPrefix(migrations, [ + { name: '0001_init.sql' }, + { name: '0002_projects.sql' }, + { name: '0003_unknown.sql' }, + ]), + ).toThrow('only 2 local migrations exist') + }) + + it('sorts safe migration files and rejects unsafe ledger names', async () => { + const directory = await mkdtemp(join(tmpdir(), 'spool-d1-list-test-')) + try { + await writeFile(join(directory, '0002_second.sql'), 'SELECT 2;') + await writeFile(join(directory, '0001_first.sql'), 'SELECT 1;') + await writeFile(join(directory, 'README.md'), 'ignored') + await expect(listMigrationNames(directory)).resolves.toEqual([ + '0001_first.sql', + '0002_second.sql', + ]) + + await writeFile(join(directory, "0003_bad'name.sql"), 'SELECT 3;') + await expect(listMigrationNames(directory)).rejects.toThrow('Migration filename is not safe') + } finally { + await rm(directory, { force: true, recursive: true }) + } + }) + + it('imports pending migrations with the ledger in the same file and verifies the result', async () => { + const directory = await mkdtemp(join(tmpdir(), 'spool-d1-runner-test-')) + const ledger: Array<{ id: number; name: string }> = [{ id: 1, name: '0001_first.sql' }] + const executeCommand = vi.fn(async () => [{ results: [], success: true }]) + const executeImport = vi.fn(async (_options: unknown, file: string) => { + const imported = await readFile(file, 'utf8') + expect(imported).toContain('CREATE TABLE projects') + expect(imported).toContain('SELECT 1, COUNT(*) FROM "d1_migrations";') + expect(imported).toContain(`INSERT INTO "d1_migrations" (name) VALUES ('0002_projects.sql');`) + ledger.push({ id: 2, name: '0002_projects.sql' }) + }) + const readLedger = vi.fn(async () => [...ledger]) + + try { + await writeFile(join(directory, '0001_first.sql'), 'CREATE TABLE first(id INTEGER);') + await writeFile(join(directory, '0002_projects.sql'), 'CREATE TABLE projects(id INTEGER);') + + await migrateRemote( + { + config: 'wrangler.test.toml', + database: 'test-db', + migrationsDir: directory, + }, + { executeCommand, executeImport, readLedger }, + ) + + expect(executeCommand).toHaveBeenCalledWith( + expect.objectContaining({ database: 'test-db' }), + MIGRATIONS_TABLE_SQL, + ) + expect(executeImport).toHaveBeenCalledTimes(1) + expect(readLedger).toHaveBeenCalled() + } finally { + await rm(directory, { force: true, recursive: true }) + } + }) + + it('fails closed when an import fails without a matching committed ledger row', async () => { + const directory = await mkdtemp(join(tmpdir(), 'spool-d1-failure-test-')) + const failure = new Error('remote import failed') + try { + await writeFile(join(directory, '0001_first.sql'), 'CREATE TABLE first(id INTEGER);') + await expect( + migrateRemote( + { + config: 'wrangler.test.toml', + database: 'test-db', + migrationsDir: directory, + }, + { + executeCommand: vi.fn(async () => [{ results: [], success: true }]), + executeImport: vi.fn(async () => { + throw failure + }), + readLedger: vi.fn(async () => []), + }, + ), + ).rejects.toBe(failure) + } finally { + await rm(directory, { force: true, recursive: true }) + } + }) + + it('does not import when the ledger is already current', async () => { + const directory = await mkdtemp(join(tmpdir(), 'spool-d1-current-test-')) + const executeImport = vi.fn() + try { + await writeFile(join(directory, '0001_first.sql'), 'CREATE TABLE first(id INTEGER);') + await migrateRemote( + { + config: 'wrangler.test.toml', + database: 'test-db', + migrationsDir: directory, + }, + { + executeCommand: vi.fn(async () => [{ results: [], success: true }]), + executeImport, + readLedger: vi.fn(async () => [{ id: 1, name: '0001_first.sql' }]), + }, + ) + expect(executeImport).not.toHaveBeenCalled() + } finally { + await rm(directory, { force: true, recursive: true }) + } + }) + + it('reconciles an ambiguous failure when the ledger proves the import committed', async () => { + const directory = await mkdtemp(join(tmpdir(), 'spool-d1-reconcile-test-')) + const ledger: Array<{ id: number; name: string }> = [] + try { + await writeFile(join(directory, '0001_first.sql'), 'CREATE TABLE first(id INTEGER);') + await expect( + migrateRemote( + { + config: 'wrangler.test.toml', + database: 'test-db', + migrationsDir: directory, + }, + { + executeCommand: vi.fn(async () => [{ results: [], success: true }]), + executeImport: vi.fn(async () => { + ledger.push({ id: 1, name: '0001_first.sql' }) + throw new Error('connection closed after commit') + }), + readLedger: vi.fn(async () => [...ledger]), + }, + ), + ).resolves.toBeUndefined() + } finally { + await rm(directory, { force: true, recursive: true }) + } + }) +}) From 3408dcde9b8bbf6255998f7abed32c96784c53ea Mon Sep 17 00:00:00 2001 From: "Yifeng \"Evan\" Wang" <7312949+doodlewind@users.noreply.github.com> Date: Sun, 26 Jul 2026 11:20:13 +0800 Subject: [PATCH 2/2] test: avoid worker URL type collision --- apps/backend/tests/d1-migrate-remote.test.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/apps/backend/tests/d1-migrate-remote.test.ts b/apps/backend/tests/d1-migrate-remote.test.ts index d07f941a..3ca37b33 100644 --- a/apps/backend/tests/d1-migrate-remote.test.ts +++ b/apps/backend/tests/d1-migrate-remote.test.ts @@ -1,6 +1,7 @@ import { mkdtemp, readFile, rm, writeFile } from 'node:fs/promises' import { tmpdir } from 'node:os' -import { join } from 'node:path' +import { dirname, join } from 'node:path' +import { fileURLToPath } from 'node:url' import { describe, expect, it, vi } from 'vite-plus/test' @@ -31,8 +32,9 @@ describe('remote D1 migration runner', () => { }) it('keeps production and staging on the atomic remote runner', async () => { + const testDirectory = dirname(fileURLToPath(import.meta.url)) const packageJson = JSON.parse( - await readFile(new URL('../package.json', import.meta.url), 'utf8'), + await readFile(join(testDirectory, '..', 'package.json'), 'utf8'), ) as { scripts: Record }