graphlemon: merge e66b3bef-f6c6-47e0-8cfe-3c9bb7fd988d

graphlemon
2 parents 44fe7b3f c3a5815a
Showing 2 changed files with 207 additions and 9 deletions
supervisor/deploy_stage.py
supervisor/test_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)
@@ -9,6 +9,7 @@ Run: python3 -m unittest discover -s supervisor -p &quot;test_*.py&quot; -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-