diff --git a/supervisor/deploy_stage.py b/supervisor/deploy_stage.py index 9c1803a..182379a 100644 --- a/supervisor/deploy_stage.py +++ b/supervisor/deploy_stage.py @@ -28,6 +28,7 @@ staging and cutover would otherwise lose track of it entirely. import os import shutil import socket +import sqlite3 import subprocess import sys import uuid @@ -42,6 +43,34 @@ class CheckoutFailedError(Exception): time this is raised.""" +class ConcurrentDeploymentError(Exception): + """CONCURRENT_DEPLOYMENT -- ux_deployment_active unique index detected an + existing deployment row not yet in a terminal state (finalized/ + rolled_back/blocked_pending_human) when attempting to INSERT a new one. + Unlike CheckoutFailedError, the INSERT never committed, so there is no + half-finished deployment row to mark blocked_pending_human -- callers + (the Deploy Supervisor polling loop) should treat this as 'skip this + round, retry next poll', not as a failure needing human intervention. + Diagnostic fields are real attributes, not just baked into the message: + - git_commit_hash: the commit hash this rejected call attempted to stage. + - blocking_deployment_id / blocking_status: the deployment row currently + holding the slot (if it could be located via a best-effort read-only + query); both None if it could not be located -- the exception is + still raised in that case. + """ + + def __init__(self, git_commit_hash, blocking_deployment_id=None, blocking_status=None): + self.git_commit_hash = git_commit_hash + self.blocking_deployment_id = blocking_deployment_id + self.blocking_status = blocking_status + message = "cannot stage commit {}: a deployment is already in progress".format(git_commit_hash) + if blocking_deployment_id is not None: + message += " (blocking deployment_id={}, status={})".format( + blocking_deployment_id, blocking_status + ) + super().__init__(message) + + def _last_finalized_deployment(conn): """Returns (git_commit_hash, candidate_instance_dir) of the most recent `finalized` deployment, or None if there isn't one yet (the bootstrap @@ -145,6 +174,37 @@ def _launch_candidate(candidate_dir, port, db_path, python_executable=None): ) +def _is_concurrent_deployment_conflict(error): + """Distinguish a ux_deployment_active unique-index conflict from any + other IntegrityError. sqlite reports named UNIQUE INDEX conflicts with + the index name embedded in the message, e.g. "UNIQUE constraint + failed: index 'ux_deployment_active'". Match only this exact index + name string -- do not broadly catch all IntegrityErrors, or unrelated + integrity errors would be misclassified and masked as concurrency + conflicts. + """ + return "ux_deployment_active" in str(error) + + +def _find_blocking_deployment(conn): + """Read-only diagnostic query: find the non-terminal deployment row + currently occupying the ux_deployment_active slot. Only called after a + conflict has already been detected, purely to enrich the exception. If + the query itself fails (e.g. a broken connection), swallow the + sqlite3.Error and return None -- the caller still raises + ConcurrentDeploymentError, just without blocking_deployment_id/status. + """ + try: + row = conn.execute( + "SELECT deployment_id, status FROM deployment " + "WHERE status NOT IN ('finalized', 'rolled_back', 'blocked_pending_human') " + "ORDER BY rowid DESC LIMIT 1" + ).fetchone() + return tuple(row) if row else None + except sqlite3.Error: + return None + + def stage_deployment( conn, git_commit_hash, repo_root=None, prod_db_path=None, checkout_parent_dir=None, port=None, python_executable=None, @@ -167,13 +227,25 @@ def stage_deployment( previous_commit_hash = _current_commit(repo_root) previous_instance_dir = repo_root - conn.execute( - "INSERT INTO deployment " - "(deployment_id, git_commit_hash, previous_commit_hash, status, previous_instance_dir, started_at) " - "VALUES (?, ?, ?, 'staged', ?, CURRENT_TIMESTAMP)", - (deployment_id, git_commit_hash, previous_commit_hash, previous_instance_dir), - ) - conn.commit() + try: + conn.execute( + "INSERT INTO deployment " + "(deployment_id, git_commit_hash, previous_commit_hash, status, previous_instance_dir, started_at) " + "VALUES (?, ?, ?, 'staged', ?, CURRENT_TIMESTAMP)", + (deployment_id, git_commit_hash, previous_commit_hash, previous_instance_dir), + ) + conn.commit() + except sqlite3.IntegrityError as e: + conn.rollback() + if not _is_concurrent_deployment_conflict(e): + raise + blocking = _find_blocking_deployment(conn) + blocking_deployment_id, blocking_status = blocking if blocking else (None, None) + raise ConcurrentDeploymentError( + git_commit_hash=git_commit_hash, + blocking_deployment_id=blocking_deployment_id, + blocking_status=blocking_status, + ) from e try: candidate_dir = _checkout_candidate(repo_root, git_commit_hash, deployment_id, checkout_parent_dir) diff --git a/supervisor/test_deploy_stage.py b/supervisor/test_deploy_stage.py index 57ed7f6..a13f5d4 100644 --- a/supervisor/test_deploy_stage.py +++ b/supervisor/test_deploy_stage.py @@ -9,6 +9,7 @@ Run: python3 -m unittest discover -s supervisor -p "test_*.py" -v import os import shutil +import sqlite3 import subprocess import sys import tempfile @@ -171,13 +172,19 @@ class StageDeploymentHappyPathTest(DeployStageTestBase): def test_only_one_non_terminal_deployment_allowed(self): with patch.object(deploy_stage, "_launch_candidate") as mock_launch: mock_launch.return_value.pid = 1 - deploy_stage.stage_deployment( + first = deploy_stage.stage_deployment( self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9001, ) - with self.assertRaises(Exception): # schema's ux_deployment_active partial unique index + with self.assertRaises(deploy_stage.ConcurrentDeploymentError) as ctx: deploy_stage.stage_deployment( self.conn, self.commit_v1, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9002, ) + err = ctx.exception + self.assertEqual(err.git_commit_hash, self.commit_v1) + self.assertEqual(err.blocking_deployment_id, first["deployment_id"]) + self.assertEqual(err.blocking_status, "candidate_starting") + count = self.conn.execute("SELECT COUNT(*) FROM deployment").fetchone()[0] + self.assertEqual(count, 1) class CheckoutFailedTest(DeployStageTestBase): @@ -198,6 +205,125 @@ class CheckoutFailedTest(DeployStageTestBase): mock_launch.assert_not_called() +class ConcurrentDeploymentTest(DeployStageTestBase): + """Tests for `ConcurrentDeploymentError` -- the diagnosable replacement + for the raw `sqlite3.IntegrityError` `ux_deployment_active` used to let + escape uncaught.""" + + # Every deployment.status value from src/schema.py's CHECK constraint + # that is NOT in ux_deployment_active's exclusion set (finalized/ + # rolled_back/blocked_pending_human) -- each of these must block a new + # stage_deployment() call, including the less-obvious terminal-looking + # ones (health_check_failed, rollback_failed) that are NOT actually + # excluded by the index. + _NON_TERMINAL_STATUSES = [ + "staged", "candidate_starting", "health_check_running", "health_check_passed", + "health_check_failed", "cutover_in_progress", "live", "monitoring", + "anomaly_detected", "rollback_in_progress", "rollback_failed", "monitoring_window_clean", + ] + + def _seed_deployment_row(self, deployment_id, status): + self.conn.execute( + "INSERT INTO deployment (deployment_id, git_commit_hash, previous_commit_hash, status, " + "candidate_instance_dir, previous_instance_dir) " + "VALUES (?, 'seed-commit', 'seed-previous-commit', ?, '/seed/candidate/dir', '/seed/previous/dir')", + (deployment_id, status), + ) + self.conn.commit() + + def test_rolled_back_prior_deployment_allows_new_staging(self): + self._seed_deployment_row("dep-rolled-back", "rolled_back") + with patch.object(deploy_stage, "_launch_candidate") as mock_launch: + mock_launch.return_value.pid = 1 + result = deploy_stage.stage_deployment( + self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9010, + ) + self.assertEqual(result["status"], "candidate_starting") + row = self.conn.execute( + "SELECT status FROM deployment WHERE deployment_id = ?", (result["deployment_id"],) + ).fetchone() + self.assertEqual(row[0], "candidate_starting") + + def test_blocked_pending_human_prior_deployment_allows_new_staging(self): + self._seed_deployment_row("dep-blocked", "blocked_pending_human") + with patch.object(deploy_stage, "_launch_candidate") as mock_launch: + mock_launch.return_value.pid = 1 + result = deploy_stage.stage_deployment( + self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9011, + ) + self.assertEqual(result["status"], "candidate_starting") + row = self.conn.execute( + "SELECT status FROM deployment WHERE deployment_id = ?", (result["deployment_id"],) + ).fetchone() + self.assertEqual(row[0], "candidate_starting") + + def test_is_concurrent_deployment_conflict_matches_only_the_exact_index_name(self): + self.assertTrue( + deploy_stage._is_concurrent_deployment_conflict( + sqlite3.IntegrityError("UNIQUE constraint failed: index 'ux_deployment_active'") + ) + ) + self.assertFalse( + deploy_stage._is_concurrent_deployment_conflict( + sqlite3.IntegrityError("NOT NULL constraint failed: deployment.git_commit_hash") + ) + ) + self.assertFalse( + deploy_stage._is_concurrent_deployment_conflict( + sqlite3.IntegrityError("UNIQUE constraint failed: index 'ux_some_other_index'") + ) + ) + + def test_find_blocking_deployment_returns_most_recently_inserted_by_rowid(self): + # ux_deployment_active itself forbids two non-terminal rows from + # coexisting (that's the whole point of this fix) -- drop it just + # for this test so two can be seeded directly, to isolate and prove + # _find_blocking_deployment's own "most recent by rowid" ordering + # logic independent of that constraint. + self.conn.execute("DROP INDEX ux_deployment_active") + self._seed_deployment_row("dep-older", "staged") + self._seed_deployment_row("dep-newer", "candidate_starting") + blocking = deploy_stage._find_blocking_deployment(self.conn) + self.assertEqual(blocking, ("dep-newer", "candidate_starting")) + + def test_find_blocking_deployment_returns_none_on_broken_connection(self): + broken_conn = schema.connect(":memory:") + broken_conn.close() + self.assertIsNone(deploy_stage._find_blocking_deployment(broken_conn)) + + def test_conn_still_usable_after_conflict_is_caught(self): + with patch.object(deploy_stage, "_launch_candidate") as mock_launch: + mock_launch.return_value.pid = 1 + deploy_stage.stage_deployment( + self.conn, self.commit_v2, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9012, + ) + with self.assertRaises(deploy_stage.ConcurrentDeploymentError): + deploy_stage.stage_deployment( + self.conn, self.commit_v1, repo_root=self.repo_dir, prod_db_path=self.prod_db_path, port=9013, + ) + # conn.rollback() in the except branch must clear the dangling + # implicit transaction -- a subsequent trivial query must not raise. + self.assertEqual(self.conn.execute("SELECT 1").fetchone()[0], 1) + + def test_every_non_excluded_status_blocks_staging_with_concurrent_deployment_error(self): + with patch.object(deploy_stage, "_launch_candidate") as mock_launch: + for status in self._NON_TERMINAL_STATUSES: + deployment_id = "dep-status-{}".format(status) + self._seed_deployment_row(deployment_id, status) + with self.assertRaises(deploy_stage.ConcurrentDeploymentError, msg=status) as ctx: + deploy_stage.stage_deployment( + self.conn, self.commit_v2, repo_root=self.repo_dir, + prod_db_path=self.prod_db_path, port=9020, + ) + self.assertEqual(ctx.exception.blocking_deployment_id, deployment_id, status) + self.assertEqual(ctx.exception.blocking_status, status, status) + mock_launch.assert_not_called() + count = self.conn.execute("SELECT COUNT(*) FROM deployment").fetchone()[0] + self.assertEqual(count, 1, status) + self.conn.execute("DELETE FROM deployment WHERE deployment_id = ?", (deployment_id,)) + self.conn.commit() + + class LaunchCandidateArgsTest(unittest.TestCase): """2026-08-06 (task_reconciliation): `_launch_candidate`'s `db_path` is always a COPY of the real production db -- without `--disable-