diff --git a/control_plane_db.py b/control_plane_db.py index 1e70a06..cbc09fa 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -318,33 +318,75 @@ class ControlPlaneDB: conn.close() def _migrate_incident_links_null_scope(self, conn: sqlite3.Connection) -> None: - """Coerce NULL scope columns to '' and collapse any duplicate keys (#619 RC).""" + """Collapse legacy NULL-scope duplicates, then normalize to '' (#619 RC2). + + Order is mandatory: SQLite UNIQUE treats multiple NULLs as distinct, so + normalizing NULL→'' first can hit the UNIQUE constraint and abort + migration. Deduplicate under the *normalized* key first, fail closed if + Gitea targets conflict within a group, then coerce NULLs to ''. + """ cols = { row[1] for row in conn.execute("PRAGMA table_info(incident_links)").fetchall() } if not cols: return + + rows = conn.execute( + """ + SELECT link_id, provider, + IFNULL(provider_base_url, '') AS base_url, + IFNULL(provider_org, '') AS p_org, + IFNULL(provider_project, '') AS p_project, + provider_issue_id, + gitea_org, gitea_repo, gitea_issue_number + FROM incident_links + ORDER BY link_id ASC + """ + ).fetchall() + + groups: dict[tuple[str, str, str, str, str], list[sqlite3.Row]] = {} + for row in rows: + key = ( + str(row["provider"]), + str(row["base_url"]), + str(row["p_org"]), + str(row["p_project"]), + str(row["provider_issue_id"]), + ) + groups.setdefault(key, []).append(row) + + to_delete: list[int] = [] + for key, members in groups.items(): + if len(members) < 2: + continue + # Canonical identity for observation links: Gitea issue target. + targets = { + ( + str(m["gitea_org"] or ""), + str(m["gitea_repo"] or ""), + int(m["gitea_issue_number"]), + ) + for m in members + } + if len(targets) > 1: + raise ControlPlaneError( + "incident_links migration fail closed: conflicting Gitea " + f"targets for provider key {key!r}: {sorted(targets)}" + ) + keep_id = int(members[0]["link_id"]) # lowest link_id (ORDER BY) + for m in members[1:]: + to_delete.append(int(m["link_id"])) + + for link_id in to_delete: + conn.execute("DELETE FROM incident_links WHERE link_id = ?", (link_id,)) + + # Safe only after dedupe: normalize NULL scope fields to empty string. for col in ("provider_base_url", "provider_org", "provider_project"): if col in cols: conn.execute( f"UPDATE incident_links SET {col} = '' WHERE {col} IS NULL" ) - # Drop duplicates keeping lowest link_id (canonical row). - conn.execute( - """ - DELETE FROM incident_links - WHERE link_id NOT IN ( - SELECT MIN(link_id) - FROM incident_links - GROUP BY provider, - IFNULL(provider_base_url, ''), - IFNULL(provider_org, ''), - IFNULL(provider_project, ''), - provider_issue_id - ) - """ - ) # ── sessions ────────────────────────────────────────────────────────── @@ -481,6 +523,11 @@ class ControlPlaneDB: raise InvalidWorkKindError( f"cannot assign kind '{kind}'; only {sorted(WORK_KINDS)}" ) + head_pin = (expected_head_sha or "").strip() + if kind_norm == "pr" and not head_pin: + raise LeaseRequiredError( + f"pr#{int(number)} assignment requires non-empty expected_head_sha pin" + ) allowed = tuple(allowed_actions or ("implement", "comment")) forbidden = tuple( @@ -825,9 +872,15 @@ class ControlPlaneDB: exp = _parse_ts(asn["expires_at"]) if exp and exp <= _utc_now(): raise LeaseRequiredError(f"lease {asn['lease_id']} expired") - # Stale-head: pinned expected_head_sha must still match live work item. + # Head pin: PRs require a non-empty expected_head_sha (fail closed). assigned_head = (asn["expected_head_sha"] or "").strip() current_head = (work["current_head_sha"] or "").strip() + if kind_norm == "pr" and not assigned_head: + raise LeaseRequiredError( + f"assignment {asn['assignment_id']} for pr#{number} has no " + f"expected_head_sha pin; PR mutations require a head pin" + ) + # Stale-head: pinned expected_head_sha must still match live work item. if assigned_head and assigned_head != current_head: raise LeaseRequiredError( f"assignment {asn['assignment_id']} stale head: expected " diff --git a/tests/test_control_plane_db.py b/tests/test_control_plane_db.py index 8535335..da8f0ed 100644 --- a/tests/test_control_plane_db.py +++ b/tests/test_control_plane_db.py @@ -123,6 +123,7 @@ class ControlPlaneDBTest(unittest.TestCase): repo="r", kind="pr", number=50, + expected_head_sha="head-50", ) b = self.db.assign_and_lease( session_id="s1", @@ -132,6 +133,7 @@ class ControlPlaneDBTest(unittest.TestCase): repo="r", kind="pr", number=50, + expected_head_sha="head-50", ) self.assertEqual(b.outcome, "assigned") self.assertEqual(b.lease_id, a.lease_id) @@ -241,6 +243,7 @@ class ControlPlaneDBTest(unittest.TestCase): repo="r", kind="pr", number=99, + expected_head_sha="merged-head", ) self.assertEqual(result.outcome, "no_safe_work") @@ -479,6 +482,210 @@ class ControlPlaneDBTest(unittest.TestCase): action="implement", ) + def test_pr_assignment_requires_expected_head_sha(self) -> None: + """PR assign and mutation must fail closed without a head pin.""" + self.db.upsert_session(session_id="s1", role="author") + with self.assertRaises(LeaseRequiredError) as ctx: + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=88, + expected_head_sha=None, + allowed_actions=("implement",), + ) + self.assertIn("expected_head_sha", str(ctx.exception)) + + # Legacy path: force an unpinned PR assignment into the DB, then + # populate head and prove mutation is still rejected. + import sqlite3 + + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=89, + current_head_sha=None, + ) + conn = sqlite3.connect(self.db_path) + try: + wid = conn.execute( + "SELECT work_item_id FROM work_items WHERE kind='pr' AND number=89" + ).fetchone()[0] + conn.execute( + """ + INSERT INTO sessions(session_id, role, started_at, last_heartbeat_at, status) + VALUES ('legacy', 'author', '2020-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active') + """ + ) + conn.execute( + """ + INSERT INTO leases( + lease_id, work_item_id, session_id, role, phase, + expires_at, heartbeat_at, status + ) VALUES ( + 'lease-legacy', ?, 'legacy', 'author', 'claimed', + '2099-01-01T00:00:00Z', '2020-01-01T00:00:00Z', 'active' + ) + """, + (wid,), + ) + conn.execute( + """ + INSERT INTO assignments( + assignment_id, work_item_id, session_id, lease_id, + allowed_actions, forbidden_actions, expected_head_sha, + role, status, created_at + ) VALUES ( + 'asn-legacy', ?, 'legacy', 'lease-legacy', + '["implement"]', '["merge"]', NULL, + 'author', 'active', '2020-01-01T00:00:00Z' + ) + """, + (wid,), + ) + conn.commit() + finally: + conn.close() + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=89, + current_head_sha="populated-head", + ) + with self.assertRaises(LeaseRequiredError) as ctx2: + self.db.require_valid_assignment( + session_id="legacy", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=89, + action="implement", + ) + self.assertIn("expected_head_sha pin", str(ctx2.exception)) + + def test_migrate_duplicate_null_scope_incident_links(self) -> None: + """Legacy NULL-scope duplicates must migrate without UNIQUE crash.""" + import sqlite3 + + from control_plane_db import ControlPlaneDB, ControlPlaneError + + # Build a v1-like table with nullable scope columns and insert dups + # that collapse under normalization, then open ControlPlaneDB on it. + path = os.path.join(self._tmp.name, "legacy_dups.sqlite3") + conn = sqlite3.connect(path) + try: + conn.executescript( + """ + CREATE TABLE schema_meta (key TEXT PRIMARY KEY, value TEXT NOT NULL); + CREATE TABLE incident_links ( + link_id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + provider_base_url TEXT, + provider_org TEXT, + provider_project TEXT, + provider_issue_id TEXT NOT NULL, + provider_short_id TEXT, + provider_permalink TEXT, + fingerprint TEXT, + gitea_org TEXT NOT NULL, + gitea_repo TEXT NOT NULL, + gitea_issue_number INTEGER NOT NULL, + linked_pr_numbers TEXT, + first_seen TEXT, + last_seen TEXT, + event_count INTEGER, + status TEXT NOT NULL DEFAULT 'open', + release_resolved_at TEXT, + last_sync_at TEXT, + UNIQUE ( + provider, provider_base_url, provider_org, + provider_project, provider_issue_id + ) + ); + INSERT INTO incident_links( + provider, provider_base_url, provider_org, provider_project, + provider_issue_id, gitea_org, gitea_repo, gitea_issue_number, status + ) VALUES + ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'), + ('sentry', NULL, NULL, NULL, 'dup-1', 'org', 'repo', 10, 'open'); + """ + ) + # SQLite allows two NULL-scope rows with same provider/issue under UNIQUE. + n = conn.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0] + self.assertEqual(n, 2) + conn.commit() + finally: + conn.close() + + db = ControlPlaneDB(path) + conn2 = sqlite3.connect(path) + try: + n2 = conn2.execute("SELECT COUNT(*) FROM incident_links").fetchone()[0] + rows = conn2.execute( + "SELECT provider_base_url, provider_org, provider_project, gitea_issue_number " + "FROM incident_links" + ).fetchall() + finally: + conn2.close() + self.assertEqual(n2, 1) + self.assertEqual(rows[0][0], "") + self.assertEqual(rows[0][1], "") + self.assertEqual(rows[0][2], "") + self.assertEqual(rows[0][3], 10) + # Touch to silence unused import in type checkers if needed + self.assertTrue(issubclass(ControlPlaneError, Exception)) + del db + + def test_migrate_conflicting_duplicate_incident_links_fails_closed(self) -> None: + """Conflicting Gitea targets for the same provider key must fail closed.""" + import sqlite3 + from control_plane_db import ControlPlaneDB, ControlPlaneError + + path = os.path.join(self._tmp.name, "legacy_conflict.sqlite3") + conn = sqlite3.connect(path) + try: + conn.executescript( + """ + CREATE TABLE incident_links ( + link_id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + provider_base_url TEXT, + provider_org TEXT, + provider_project TEXT, + provider_issue_id TEXT NOT NULL, + gitea_org TEXT NOT NULL, + gitea_repo TEXT NOT NULL, + gitea_issue_number INTEGER NOT NULL, + status TEXT NOT NULL DEFAULT 'open', + UNIQUE ( + provider, provider_base_url, provider_org, + provider_project, provider_issue_id + ) + ); + INSERT INTO incident_links( + provider, provider_base_url, provider_org, provider_project, + provider_issue_id, gitea_org, gitea_repo, gitea_issue_number + ) VALUES + ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 1), + ('sentry', NULL, NULL, NULL, 'dup-c', 'org', 'repo', 2); + """ + ) + conn.commit() + finally: + conn.close() + + with self.assertRaises(ControlPlaneError) as ctx: + ControlPlaneDB(path) + self.assertIn("conflicting", str(ctx.exception).lower()) + def test_incident_links_minimal_upsert_is_canonical(self) -> None: """Repeated minimal upserts must update one row (NULL-safe uniqueness).""" a = self.db.upsert_incident_link(