also looking at this
fix(migrate): apply each migration atomically, and one run at a time
#1076
3 files
+364
-54
| @@ -383,6 +383,88 @@ export async function generateMigration(dir?: string, opts: MigrateOptions = {}) | ||
| 383 | 383 | |
| 384 | 384 | // `dir` is the models directory. This function runs the SQL corpus, which it |
| 385 | 385 | // finds from the workspace root, so it has no use for it — the parameter stays |
| 386 | /** | |
| 387 | * The advisory-lock name every migrate run contends on. | |
| 388 | * | |
| 389 | * Constant on purpose: two processes must pick the same key to serialise, and | |
| 390 | * migrations are global to a database rather than per-model or per-directory. | |
| 391 | */ | |
| 392 | const MIGRATION_LOCK_KEY = 'bun-query-builder:migrate' | |
| 393 | ||
| 394 | /** | |
| 395 | * Statements that cannot run inside a transaction, so a file containing one | |
| 396 | * must be applied unwrapped. | |
| 397 | * | |
| 398 | * Postgres rejects `CREATE INDEX CONCURRENTLY` and `DROP INDEX CONCURRENTLY` | |
| 399 | * inside a transaction block outright, and `VACUUM` likewise — and building an | |
| 400 | * index concurrently is exactly what a migration on a live table SHOULD do, so | |
| 401 | * refusing to support it is not an option. Files that use them keep the old | |
| 402 | * unwrapped behaviour and say so, rather than failing on something they were | |
| 403 | * right to write. | |
| 404 | */ | |
| 405 | const NON_TRANSACTIONAL_SQL = /\bCONCURRENTLY\b|^\s*VACUUM\b/im | |
| 406 | ||
| 407 | /** | |
| 408 | * Whether this dialect can roll a failed migration file back. | |
| 409 | * | |
| 410 | * Postgres has fully transactional DDL. SQLite does too, for the statements a | |
| 411 | * migration uses. MySQL does NOT: `CREATE TABLE`, `ALTER TABLE` and friends | |
| 412 | * force an implicit commit, so wrapping a file there produces a transaction | |
| 413 | * that silently commits mid-way and hands back false confidence — worse than | |
| 414 | * not wrapping, because it looks like it worked. | |
| 415 | */ | |
| 416 | function supportsTransactionalDdl(dialect: SupportedDialect): boolean { | |
| 417 | return dialect === 'postgres' || dialect === 'sqlite' | |
| 418 | } | |
| 419 | ||
| 420 | /** | |
| 421 | * Run `fn` while holding the migration advisory lock. | |
| 422 | * | |
| 423 | * Two processes booting at once — an ordinary rolling deploy — both read the | |
| 424 | * same pending set and both apply it. The lock makes the second wait for the | |
| 425 | * first, then find nothing pending. | |
| 426 | * | |
| 427 | * The lock is taken on a RESERVED connection and released on the same one. | |
| 428 | * That is not incidental: a Postgres session lock lives until it is released | |
| 429 | * or the session ends, and a pooled connection does not end when it goes back | |
| 430 | * to the pool. Taking it on a pooled builder would strand the lock on whatever | |
| 431 | * connection happened to serve the call, and the next migrate would wait | |
| 432 | * forever on a lock nobody meant to hold. | |
| 433 | * | |
| 434 | * SQLite has no advisory-lock primitive and a single-writer file, so it runs | |
| 435 | * unlocked rather than throwing. | |
| 436 | */ | |
| 437 | async function withMigrationLock<T>(dialect: SupportedDialect, fn: () => Promise<T>): Promise<T> { | |
| 438 | if (dialect === 'sqlite') | |
| 439 | return await fn() | |
| 440 | ||
| 441 | const qb = createQueryBuilder() as any | |
| 442 | let reserved: any | |
| 443 | try { | |
| 444 | reserved = await qb.reserve() | |
| 445 | } | |
| 446 | catch { | |
| 447 | // A driver that cannot reserve (or a mock) should not block migrating. | |
| 448 | return await fn() | |
| 449 | } | |
| 450 | ||
| 451 | let held = false | |
| 452 | try { | |
| 453 | await reserved.advisoryLock(MIGRATION_LOCK_KEY) | |
| 454 | held = true | |
| 455 | return await fn() | |
| 456 | } | |
| 457 | finally { | |
| 458 | try { | |
| 459 | if (held) | |
| 460 | await reserved.advisoryUnlock(MIGRATION_LOCK_KEY) | |
| 461 | } | |
| 462 | finally { | |
| 463 | reserved.release?.() | |
| 464 | } | |
| 465 | } | |
| 466 | } | |
| 467 | ||
| 386 | 468 | // because it is public API and every caller passes it. |
| 387 | 469 | // eslint-disable-next-line pickier/no-unused-vars |
| 388 | 470 | export async function executeMigration(dir?: string): Promise<boolean> { |
| @@ -417,58 +499,86 @@ export async function executeMigration(dir?: string): Promise<boolean> { | ||
| 417 | 499 | // Create migrations table if it doesn't exist |
| 418 | 500 | await createMigrationsTable(qb, dialect) |
| 419 | 501 | |
| 420 | // Get already executed migrations | |
| 421 | const executedMigrations = await getExecutedMigrations(qb, dialect) | |
| 422 | ||
| 423 | /* | |
| 424 | * Every migration file is permanent, and every one that runs is recorded. | |
| 425 | * | |
| 426 | * Generated ALTERs used to be "transient": replayed on every migrate | |
| 427 | * rather than recorded, then deleted from disk once applied. That made | |
| 428 | * the corpus a log of one machine's session instead of the schema. Change | |
| 429 | * an attribute, migrate, and the ALTER that carried the change was gone — | |
| 430 | * so a teammate who cloned the repo and migrated got the table as it was | |
| 431 | * BEFORE the change, with no error and no clue. The generator writes each | |
| 432 | * change once (`createMigrationFile` refuses to write a statement the | |
| 433 | * corpus already contains), so there is nothing left for replay to fix, | |
| 434 | * and everything to lose by deleting the evidence. | |
| 435 | */ | |
| 436 | const pending = scriptFiles.filter(file => !executedMigrations.includes(file)) | |
| 437 | ||
| 438 | if (pending.length === 0) { | |
| 439 | info('-- No pending migrations to execute') | |
| 440 | return true | |
| 441 | } | |
| 502 | // Everything from here runs under the migration lock, and the read of the | |
| 503 | // ledger is deliberately INSIDE it: computing `pending` before locking is | |
| 504 | // what lets two processes agree on the same work and then both do it. | |
| 505 | return await withMigrationLock(dialect, async () => { | |
| 506 | // Get already executed migrations | |
| 507 | const executedMigrations = await getExecutedMigrations(qb, dialect) | |
| 508 | ||
| 509 | /* | |
| 510 | * Every migration file is permanent, and every one that runs is recorded. | |
| 511 | * | |
| 512 | * Generated ALTERs used to be "transient": replayed on every migrate | |
| 513 | * rather than recorded, then deleted from disk once applied. That made | |
| 514 | * the corpus a log of one machine's session instead of the schema. Change | |
| 515 | * an attribute, migrate, and the ALTER that carried the change was gone — | |
| 516 | * so a teammate who cloned the repo and migrated got the table as it was | |
| 517 | * BEFORE the change, with no error and no clue. The generator writes each | |
| 518 | * change once (`createMigrationFile` refuses to write a statement the | |
| 519 | * corpus already contains), so there is nothing left for replay to fix, | |
| 520 | * and everything to lose by deleting the evidence. | |
| 521 | */ | |
| 522 | const pending = scriptFiles.filter(file => !executedMigrations.includes(file)) | |
| 523 | ||
| 524 | if (pending.length === 0) { | |
| 525 | info('-- No pending migrations to execute') | |
| 526 | return true | |
| 527 | } | |
| 442 | 528 | |
| 443 | info(`-- Executing ${pending.length} migration${pending.length === 1 ? '' : 's'}`) | |
| 529 | info(`-- Executing ${pending.length} migration${pending.length === 1 ? '' : 's'}`) | |
| 444 | 530 | |
| 445 | for (const file of pending) { | |
| 446 | const filePath = join(sqlDir, file) | |
| 447 | info(`-- Executing: ${file}`) | |
| 531 | for (const file of pending) { | |
| 532 | const filePath = join(sqlDir, file) | |
| 533 | info(`-- Executing: ${file}`) | |
| 448 | 534 | |
| 449 | try { | |
| 450 | await qb.file(filePath) | |
| 451 | await recordMigration(qb, file, dialect) | |
| 452 | info(`-- ✓ Migration ${file} executed and recorded`) | |
| 453 | } | |
| 454 | catch (err) { | |
| 455 | // A generated migration whose change is already in place is not a | |
| 456 | // failure, it is a bookkeeping gap: the file was applied on a run that | |
| 457 | // died before recording it, or under the old transient behaviour that | |
| 458 | // never recorded it at all. Record it and carry on — the schema is | |
| 459 | // where the file says it should be. Anything else, including every | |
| 460 | // authored migration, fails loudly. | |
| 461 | if (isGeneratedMigration(sqlDir, file) && isAlreadyAppliedError(err)) { | |
| 462 | await recordMigration(qb, file, dialect) | |
| 463 | info(`-- ✓ Migration ${file} was already applied; recorded it (${err instanceof Error ? err.message : String(err)})`) | |
| 464 | continue | |
| 535 | try { | |
| 536 | // Apply the file and record it as ONE unit where the dialect allows. | |
| 537 | // | |
| 538 | // Unwrapped, a file is applied statement by statement: if statement 3 | |
| 539 | // throws, statements 1 and 2 are already committed and the file is | |
| 540 | // neither applied nor cleanly re-runnable. Recording inside the same | |
| 541 | // transaction closes the matching gap at the other end, where the | |
| 542 | // schema changed but the process died before the ledger was written. | |
| 543 | const wrappable = supportsTransactionalDdl(dialect) | |
| 544 | && !NON_TRANSACTIONAL_SQL.test(readFileSync(filePath, 'utf-8')) | |
| 545 | ||
| 546 | if (wrappable) { | |
| 547 | await qb.transaction(async (tx: any) => { | |
| 548 | await tx.file(filePath) | |
| 549 | await recordMigration(tx, file, dialect) | |
| 550 | }) | |
| 551 | } | |
| 552 | else { | |
| 553 | await qb.file(filePath) | |
| 554 | await recordMigration(qb, file, dialect) | |
| 555 | if (!supportsTransactionalDdl(dialect)) | |
| 556 | info(`-- (${dialect} commits DDL implicitly, so ${file} was not applied atomically)`) | |
| 557 | else | |
| 558 | info(`-- (${file} contains a statement that cannot run in a transaction, so it was not applied atomically)`) | |
| 559 | } | |
| 560 | info(`-- ✓ Migration ${file} executed and recorded`) | |
| 561 | } | |
| 562 | catch (err) { | |
| 563 | // A generated migration whose change is already in place is not a | |
| 564 | // failure, it is a bookkeeping gap: the file was applied on a run that | |
| 565 | // died before recording it, or under the old transient behaviour that | |
| 566 | // never recorded it at all. Record it and carry on — the schema is | |
| 567 | // where the file says it should be. Anything else, including every | |
| 568 | // authored migration, fails loudly. | |
| 569 | if (isGeneratedMigration(sqlDir, file) && isAlreadyAppliedError(err)) { | |
| 570 | await recordMigration(qb, file, dialect) | |
| 571 | info(`-- ✓ Migration ${file} was already applied; recorded it (${err instanceof Error ? err.message : String(err)})`) | |
| 572 | continue | |
| 573 | } | |
| 574 | console.error(`-- ✗ Migration ${file} failed:`, err) | |
| 575 | throw err | |
| 465 | 576 | } |
| 466 | console.error(`-- ✗ Migration ${file} failed:`, err) | |
| 467 | throw err | |
| 468 | 577 | } |
| 469 | } | |
| 470 | 578 | |
| 471 | info('-- All migrations executed successfully') | |
| 579 | info('-- All migrations executed successfully') | |
| 580 | return true | |
| 581 | }) | |
| 472 | 582 | } |
| 473 | 583 | catch (err) { |
| 474 | 584 | console.error('-- Migration execution failed:', err) |