From 0ade73a5684b08e71451cf91712d014ac3f4fed8 Mon Sep 17 00:00:00 2001 From: Atilio Hector Date: Sat, 20 Jun 2026 23:21:15 -0300 Subject: [PATCH 1/2] fix: pass uuid4() call result instead of function object in rollback query_id --- housewatch/async_migrations/runner.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/housewatch/async_migrations/runner.py b/housewatch/async_migrations/runner.py index d432d63..750ec3e 100644 --- a/housewatch/async_migrations/runner.py +++ b/housewatch/async_migrations/runner.py @@ -157,7 +157,7 @@ def attempt_migration_rollback(migration: AsyncMigration): op = ops[op_index] if not op: continue - execute_op(op, query_id=str(uuid4)) + execute_op(op, query_id=str(uuid4())) except Exception as e: last_error = f"At operation {op_index} rollback failed with error:{str(e)}" process_error( From 6058147ddd37c09f90d6f97a77d91a44822c0e6b Mon Sep 17 00:00:00 2001 From: Atilio Hector Date: Sun, 21 Jun 2026 17:53:33 -0300 Subject: [PATCH 2/2] fix: enforce single concurrent migration with atomic cross-row lock --- .../async_migrations/async_migration_utils.py | 39 ++++++++++++++----- housewatch/async_migrations/runner.py | 5 ++- 2 files changed, 34 insertions(+), 10 deletions(-) diff --git a/housewatch/async_migrations/async_migration_utils.py b/housewatch/async_migrations/async_migration_utils.py index 5483011..89ff62a 100644 --- a/housewatch/async_migrations/async_migration_utils.py +++ b/housewatch/async_migrations/async_migration_utils.py @@ -29,17 +29,38 @@ def execute_op(sql: str, args=None, *, query_id: str, timeout_seconds: int = 600 def mark_async_migration_as_running(migration: AsyncMigration) -> bool: - # update to running iff the state was Starting (ui triggered) or NotStarted (api triggered) with transaction.atomic(): - if migration.status not in [MigrationStatus.Starting, MigrationStatus.NotStarted]: + # Cross-row check: lock all rows in Running or Starting state before proceeding. + # This blocks any concurrent transaction attempting the same check, ensuring only + # one migration transitions to Running at a time across the entire table. + already_running = ( + AsyncMigration.objects + .select_for_update() + .filter(status__in=[MigrationStatus.Running, MigrationStatus.Starting]) + .exclude(pk=migration.pk) + .exists() + ) + if already_running: + logger.warning( + "Refusing to start migration: another migration is already running or starting", + migration=migration.name, + ) + return False + + # Re-read this migration's own row under a lock to guard against double-start + # of the same migration by two concurrent workers. + instance = AsyncMigration.objects.select_for_update().get(pk=migration.pk) + if instance.status not in [MigrationStatus.Starting, MigrationStatus.NotStarted]: return False - migration.status = MigrationStatus.Running - migration.current_query_id = "" - migration.progress = 0 - migration.current_operation_index = 0 - migration.started_at = now() - migration.finished_at = None - migration.save() + + instance.status = MigrationStatus.Running + instance.current_query_id = "" + instance.progress = 0 + instance.current_operation_index = 0 + instance.started_at = now() + instance.finished_at = None + instance.save() + return True diff --git a/housewatch/async_migrations/runner.py b/housewatch/async_migrations/runner.py index 750ec3e..8699e7e 100644 --- a/housewatch/async_migrations/runner.py +++ b/housewatch/async_migrations/runner.py @@ -48,7 +48,10 @@ def start_async_migration(migration: AsyncMigration, ignore_posthog_version=Fals if not mark_async_migration_as_running(migration): # we don't want to touch the migration, i.e. don't process_error - logger.error(f"Migration state has unexpectedly changed for async migration {migration.name}") + logger.error( + "Could not start migration: another migration is already running or migration state changed unexpectedly", + migration=migration.name, + ) return False return run_async_migration_operations(migration)