Make migration:run safe to run repeatedly and in parallel

Previously a killed `migration:run` left no trace, so the next run would
silently re-execute a possibly partially-applied migration, and two runs in
parallel could both execute the same migration. This makes the command safe in
both situations (safe = never silently corrupts the database).

Two mechanisms, modeled on nextras/migrations:

1. Record the start before executing. `finished_at` is now nullable; the start
   marker (finished_at = NULL) is committed *before* the migration body and
   *outside* any transaction (only the body is wrapped in a transaction for
   TransactionalMigration). A migration interrupted mid-run therefore leaves a
   durable finished_at = NULL row that survives a body rollback (PostgreSQL) and
   a DDL implicit commit (MySQL). Every subsequent `migration:run` then fails
   loudly until the situation is resolved manually, instead of re-running
   potentially non-idempotent SQL.

2. Serialize runs with a database lock. `migration:run` acquires a lock before
   reading state and executing, releasing it in a finally block:
   - MySQL/MariaDB: GET_LOCK / RELEASE_LOCK
   - PostgreSQL: pg_try_advisory_lock / pg_advisory_unlock
   - SQLite/other: no-op (single-writer)
   The lock wait timeout is configurable (`lock_timeout_seconds`, default 300).

Also:
- `migration:init` idempotently upgrades an existing table by making
  `finished_at` nullable (narrow single-column ALTER, no full schema diff).
- `migration:run` exits 1 (unfinished migration), 2 (lock not acquired) or
  3 (table missing/not upgraded); `migration:check` reports unfinished
  migrations (exit bit 8).
- `getExecutedVersions()` now honors a custom migration table name (was
  hardcoded), and identifiers in raw SELECTs are quoted.

BC: `finished_at` becomes nullable and the incomplete-migration check can now
block runs. Run `migration:init` once after upgrading to migrate the table.

Co-Authored-By: Claude Code

Author: JanTvrdik

README.md

@@ -39,6 +39,7 @@ two_phase_migrations:
     # excluded_tables: ['my_tmp_table']
     # template_file_path: '%kernel.project_dir%/migrations/my-template.txt'
     # template_indent: "\t\t"
+    # lock_timeout_seconds: 300 # how long migration:run waits for a lock held by a parallel run before failing
 ```
 
 The bundle requires `symfony/http-kernel` ^6.4+.
@@ -74,6 +75,7 @@ services:
         $excludedTables: ['my_tmp_table'] # migration table ($migrationTableName) is always added to excluded tables automatically
         $templateFilePath: "%kernel.project_dir%/migrations/my-template.txt" # customizable according to your coding style
         $templateIndent: "\t\t" # defaults to spaces
+        $lockTimeoutSeconds: 300 # how long migration:run waits for a lock held by a parallel run before failing
 ```
 </details>
 
@@ -82,6 +84,7 @@ services:
 #### Initialization:
 
 After installation, you need to create `migration` table in your database. It is safe to run it even when the table was already initialized.
+When upgrading from `1.x`, running it once also upgrades the existing table to the new schema (it makes `finished_at` nullable, see [Concurrency & execution safety](#concurrency--execution-safety)). Only that single column is altered, so it is safe to run against a populated table.
 
 ```bash
 $ bin/console migration:init
@@ -213,6 +216,41 @@ $ bin/console migration:run both
 [info] Migration execution completed (phase both)
 ```
 
+### Concurrency & execution safety
+
+`migration:run` is designed to be **safe to run multiple times and in parallel** (e.g. during a rolling deployment where several instances may start the command at once). "Safe" here means it will never silently corrupt your database — not that every invocation finishes the work. Two mechanisms provide this:
+
+#### Parallel runs are serialized by a database lock
+
+Before doing anything, `migration:run` acquires a database-level lock, so only one run executes migrations at a time:
+
+- **MySQL / MariaDB** – a named lock via `GET_LOCK()` / `RELEASE_LOCK()`
+- **PostgreSQL** – a session-level advisory lock via `pg_try_advisory_lock()` / `pg_advisory_unlock()`
+- **other platforms (incl. SQLite)** – no-op (SQLite already serializes writers at the filesystem level)
+
+A parallel run waits up to `lock_timeout_seconds` (default `300`) for the lock. If the lock holder finishes in time, the waiting run simply proceeds and finds nothing to do (exit code `0`). If the timeout is exceeded, it aborts with exit code `2` instead of running concurrently. The lock is bound to the connection session, so it is released automatically if a runner is killed.
+
+#### Interrupted migrations block all further runs until resolved
+
+Each migration is recorded in two steps: a row with `started_at` (and `finished_at = NULL`) is committed **before** the migration body runs, and `finished_at` is set **after** it succeeds. If a runner is killed mid-migration (on MySQL, DDL such as `ALTER` auto-commits and cannot be rolled back), a `finished_at = NULL` row is left behind.
+
+Because such a migration may be only partially applied, **every subsequent `migration:run` fails loudly** (exit code `1`) instead of re-executing potentially non-idempotent SQL:
+
+```bash
+$ bin/console migration:run before
+
+# example output:
+[info] Starting migration execution (phase before)
+[error] Migration execution aborted, found 1 unfinished migration(s) from a previously interrupted run, manual resolution is required
+```
+
+To recover, inspect the database to determine what the interrupted migration actually applied, then resolve the `finished_at = NULL` row in the migration table manually:
+
+- delete the row to re-run the migration from scratch (only safe if nothing was applied or the migration is idempotent), or
+- set its `finished_at` to mark it as completed (if you verified everything was applied or finished it by hand).
+
+`migration:run` exit codes: `0` success, `1` unfinished migration found (manual resolution required), `2` lock could not be acquired, `3` migration table missing or not upgraded (run `migration:init`). `migration:check` also reports unfinished migrations (and adds `8` to its exit code bitmask).
+
 ### Advanced usage
 
 #### Run custom code for each executed query:
@@ -249,6 +287,8 @@ But those columns are declared as VARCHARs by default, because there is [no micr
 That may complicate datetime manipulations (like duration calculation).
 You can adjust the structure to your needs (e.g. use `DATETIME(6)` for MySQL) manually in some migration.
 
+`finished_at` is nullable: a `NULL` value marks a migration that was started but never finished (see [Concurrency & execution safety](#concurrency--execution-safety)).
+
 ```
 +----------------+--------+-----------------------------+---------------------------+
 | version        | phase  | started_at                  | finished_at               |

src/Bridge/Symfony/TwoPhaseMigrationsBundle.php

@@ -32,6 +32,7 @@ public function configure(DefinitionConfigurator $definition): void
             ->end()
             ->scalarNode('template_file_path')->defaultNull()->end()
             ->scalarNode('template_indent')->defaultNull()->end()
+            ->integerNode('lock_timeout_seconds')->defaultNull()->end()
             ->end();
     }
 
@@ -44,6 +45,7 @@ public function configure(DefinitionConfigurator $definition): void
      *     excluded_tables: list<string>,
      *     template_file_path: ?string,
      *     template_indent: ?string,
+     *     lock_timeout_seconds: ?int,
      * } $config
      */
     public function loadExtension( // @phpstan-ignore method.childParameterType, method.childParameterType
@@ -63,6 +65,7 @@ public function loadExtension( // @phpstan-ignore method.childParameterType, met
                 '$excludedTables' => $config['excluded_tables'],
                 '$templateFilePath' => $config['template_file_path'],
                 '$templateIndent' => $config['template_indent'],
+                '$lockTimeoutSeconds' => $config['lock_timeout_seconds'],
             ]);
 
         $services->set(MigrationService::class)

src/Command/MigrationCheckCommand.php

@@ -10,6 +10,7 @@
 use Symfony\Component\Console\Input\InputInterface;
 use Symfony\Component\Console\Output\OutputInterface;
 use function array_diff;
+use function array_map;
 use function array_values;
 use function count;
 use function implode;
@@ -22,6 +23,7 @@ class MigrationCheckCommand extends Command
 
     public const NAME = 'migration:check';
 
+    public const EXIT_INCOMPLETE_MIGRATION = 8;
     public const EXIT_ENTITIES_NOT_SYNCED = 4;
     public const EXIT_UNKNOWN_MIGRATION = 2;
     public const EXIT_AWAITING_MIGRATION = 1;
@@ -45,6 +47,7 @@ public function execute(
         $logger->info('Starting migration check');
 
         $exitCode = self::EXIT_OK;
+        $exitCode |= $this->checkIncompleteMigrations($logger);
         $exitCode |= $this->checkMigrationsExecuted($logger);
         $exitCode |= $this->checkEntitiesSyncedWithDatabase($logger);
 
@@ -56,6 +59,26 @@ public function execute(
         return $exitCode;
     }
 
+    private function checkIncompleteMigrations(LoggerInterface $logger): int
+    {
+        $incomplete = $this->migrationService->getIncompleteMigrations();
+
+        if (count($incomplete) === 0) {
+            return self::EXIT_OK;
+        }
+
+        $logger->error('Found {migrationIncompleteCount} unfinished migration(s) from a previously interrupted run, manual resolution is required: {migrationIncompleteList}', [
+            'migrationIncompleteCount' => count($incomplete),
+            'migrationIncomplete' => $incomplete,
+            'migrationIncompleteList' => implode(', ', array_map(
+                static fn (array $migration): string => $migration['version'] . ' (phase ' . $migration['phase'] . ')',
+                $incomplete,
+            )),
+        ]);
+
+        return self::EXIT_INCOMPLETE_MIGRATION;
+    }
+
     private function checkEntitiesSyncedWithDatabase(LoggerInterface $logger): int
     {
         $updates = $this->migrationService->generateDiffSqls();

src/Command/MigrationRunCommand.php

@@ -4,8 +4,11 @@
 
 use LogicException;
 use Psr\Log\LoggerInterface;
+use ShipMonk\Doctrine\Migration\IncompleteMigrationException;
+use ShipMonk\Doctrine\Migration\MigrationLockException;
 use ShipMonk\Doctrine\Migration\MigrationPhase;
 use ShipMonk\Doctrine\Migration\MigrationService;
+use ShipMonk\Doctrine\Migration\MigrationTableNotInitializedException;
 use Symfony\Component\Console\Attribute\AsCommand;
 use Symfony\Component\Console\Command\Command;
 use Symfony\Component\Console\Input\InputArgument;
@@ -28,6 +31,11 @@ class MigrationRunCommand extends Command
     public const ARGUMENT_PHASE = 'phase';
     public const PHASE_BOTH = 'both';
 
+    public const EXIT_OK = 0;
+    public const EXIT_INCOMPLETE_MIGRATION = 1;
+    public const EXIT_LOCK_NOT_ACQUIRED = 2;
+    public const EXIT_TABLE_NOT_INITIALIZED = 3;
+
     public function __construct(
         private readonly MigrationService $migrationService,
         private readonly ?LoggerInterface $logger = null,
@@ -65,19 +73,60 @@ public function execute(
             'migrationPhases' => array_map(static fn (MigrationPhase $phase): string => $phase->value, $phases),
         ]);
 
-        $migratedSomething = $this->executeMigrations($logger, $phases);
+        // Acquiring the lock first makes the whole run serialized across processes, so that running this command
+        // multiple times in parallel is safe (only one runner executes migrations, the others find nothing to do).
+        try {
+            $this->migrationService->acquireLock();
+        } catch (MigrationLockException $e) {
+            $logger->error('Migration execution aborted, could not acquire migration lock within {migrationLockTimeoutSeconds} s (another migration run is probably in progress)', [
+                'migrationPhaseArgument' => $phaseArgument,
+                'migrationLockTimeoutSeconds' => $e->timeoutSeconds,
+            ]);
+
+            return self::EXIT_LOCK_NOT_ACQUIRED;
+        }
+
+        try {
+            // Fail with an actionable message (instead of a cryptic constraint violation) when the migration table
+            // is missing or was not upgraded via migration:init after a library upgrade.
+            $this->migrationService->assertMigrationTableUpToDate();
+
+            // A migration interrupted by a previous run leaves a partially applied state that we must not silently
+            // continue from, so all subsequent runs fail loudly until the situation is resolved manually.
+            $this->migrationService->assertNoIncompleteMigrations();
+
+            $migratedSomething = $this->executeMigrations($logger, $phases);
+
+            if (!$migratedSomething) {
+                $logger->notice('No migrations to execute (phase {migrationPhaseArgument})', [
+                    'migrationPhaseArgument' => $phaseArgument,
+                ]);
+            } else {
+                $logger->info('Migration execution completed (phase {migrationPhaseArgument})', [
+                    'migrationPhaseArgument' => $phaseArgument,
+                ]);
+            }
 
-        if (!$migratedSomething) {
-            $logger->notice('No migrations to execute (phase {migrationPhaseArgument})', [
+            return self::EXIT_OK;
+        } catch (MigrationTableNotInitializedException $e) {
+            $logger->error('Migration execution aborted, migration table {migrationTableName} is not initialized, run the migration:init command first', [
                 'migrationPhaseArgument' => $phaseArgument,
+                'migrationTableName' => $e->tableName,
+                'migrationError' => $e->getMessage(),
             ]);
-        } else {
-            $logger->info('Migration execution completed (phase {migrationPhaseArgument})', [
+
+            return self::EXIT_TABLE_NOT_INITIALIZED;
+        } catch (IncompleteMigrationException $e) {
+            $logger->error('Migration execution aborted, found {migrationIncompleteCount} unfinished migration(s) from a previously interrupted run, manual resolution is required', [
                 'migrationPhaseArgument' => $phaseArgument,
+                'migrationIncompleteCount' => count($e->incompleteMigrations),
+                'migrationIncomplete' => $e->incompleteMigrations,
             ]);
-        }
 
-        return 0;
+            return self::EXIT_INCOMPLETE_MIGRATION;
+        } finally {
+            $this->migrationService->releaseLock();
+        }
     }
 
     /**

src/IncompleteMigrationException.php

@@ -0,0 +1,27 @@
+<?php declare(strict_types = 1);
+
+namespace ShipMonk\Doctrine\Migration;
+
+use RuntimeException;
+
+/**
+ * Thrown when there are migrations that were started but never finished (e.g. the process was killed mid-execution).
+ * Such migrations may be partially applied, therefore all subsequent runs must fail until the state is resolved manually.
+ *
+ * @api
+ */
+class IncompleteMigrationException extends RuntimeException
+{
+
+    /**
+     * @param list<array{version: string, phase: string, startedAt: string}> $incompleteMigrations
+     */
+    public function __construct(
+        public readonly array $incompleteMigrations,
+        string $message,
+    )
+    {
+        parent::__construct($message);
+    }
+
+}

src/MigrationConfig.php

@@ -5,10 +5,13 @@
 use LogicException;
 use function is_dir;
 use function is_file;
+use function max;
 
 class MigrationConfig
 {
 
+    private const DEFAULT_LOCK_TIMEOUT_SECONDS = 300;
+
     private string $migrationsDir;
 
     private string $migrationsTableName;
@@ -26,6 +29,8 @@ class MigrationConfig
 
     private string $templateIndent;
 
+    private int $lockTimeoutSeconds;
+
     /**
      * @param string[]|null $excludedTables
      */
@@ -37,6 +42,7 @@ public function __construct(
         ?array $excludedTables = null,
         ?string $templateFilePath = null,
         ?string $templateIndent = null,
+        ?int $lockTimeoutSeconds = null,
     )
     {
         $templateFilePathToUse = $templateFilePath ?? __DIR__ . '/template/migration.txt';
@@ -57,6 +63,7 @@ public function __construct(
         $this->excludedTables[] = $this->getMigrationTableName();
         $this->templateFilePath = $templateFilePathToUse;
         $this->templateIndent = $templateIndent ?? '        ';
+        $this->lockTimeoutSeconds = max(1, $lockTimeoutSeconds ?? self::DEFAULT_LOCK_TIMEOUT_SECONDS);
     }
 
     public function getMigrationsDirectory(): string
@@ -97,4 +104,9 @@ public function getTemplateIndent(): string
         return $this->templateIndent;
     }
 
+    public function getLockTimeoutSeconds(): int
+    {
+        return $this->lockTimeoutSeconds;
+    }
+
 }