From 783ea88a01da721138429ede8dc8a6c0c186206a Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 9 Jul 2026 22:41:25 -0400 Subject: [PATCH] fix: fail closed on stale/terminal assignments; canonicalize incident_links Address REQUEST_CHANGES on PR #619: - require_valid_assignment rejects terminal work states and head drift - incident_links scope keys normalize NULL/blank to '' for UNIQUE - remove trailing whitespace in control-plane-db-substrate.md --- control_plane_db.py | 98 +++++++++-- .../control-plane-db-substrate.md | 16 +- tests/test_control_plane_db.py | 155 +++++++++++++++++- 3 files changed, 244 insertions(+), 25 deletions(-) diff --git a/control_plane_db.py b/control_plane_db.py index adaa542..1e70a06 100644 --- a/control_plane_db.py +++ b/control_plane_db.py @@ -29,11 +29,14 @@ from dataclasses import dataclass from datetime import datetime, timedelta, timezone from typing import Any, Iterator, Sequence -SCHEMA_VERSION = 1 +SCHEMA_VERSION = 2 # Assignable work kinds only — raw monitoring incidents are never work items. WORK_KINDS = frozenset({"issue", "pr"}) +# Work item states that permanently revoke assignment/lease authority. +TERMINAL_WORK_STATES = frozenset({"merged", "closed"}) + DEFAULT_LEASE_TTL_SECONDS = 4 * 3600 # Environment: path for SQLite MVP (single-writer). @@ -120,12 +123,13 @@ CREATE TABLE IF NOT EXISTS events ( ); -- Provider-neutral incident ↔ Gitea link model (ADR §8–9). Not assignable work. +-- Optional scope fields are stored as '' (never NULL) so UNIQUE is NULL-safe. CREATE TABLE IF NOT EXISTS incident_links ( link_id INTEGER PRIMARY KEY AUTOINCREMENT, provider TEXT NOT NULL, - provider_base_url TEXT, - provider_org TEXT, - provider_project TEXT, + provider_base_url TEXT NOT NULL DEFAULT '', + provider_org TEXT NOT NULL DEFAULT '', + provider_project TEXT NOT NULL DEFAULT '', provider_issue_id TEXT NOT NULL, provider_short_id TEXT, provider_permalink TEXT, @@ -172,6 +176,17 @@ def _parse_ts(value: str | None) -> datetime | None: return None +def _norm_scope(value: str | None) -> str: + """Normalize optional incident-link scope keys for NULL-safe uniqueness. + + Empty / whitespace / None all become '' so SQLite UNIQUE treats them as one + canonical key component (SQLite treats multiple NULLs as distinct). + """ + if value is None: + return "" + return str(value).strip() + + def default_db_path() -> str: raw = (os.environ.get(DB_PATH_ENV) or DEFAULT_DB_PATH).strip() return raw or DEFAULT_DB_PATH @@ -285,6 +300,7 @@ class ControlPlaneDB: conn = self._connect() try: conn.executescript(_SCHEMA_SQL) + self._migrate_incident_links_null_scope(conn) conn.execute( "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)", ("schema_version", str(SCHEMA_VERSION)), @@ -301,6 +317,35 @@ class ControlPlaneDB: finally: 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).""" + cols = { + row[1] + for row in conn.execute("PRAGMA table_info(incident_links)").fetchall() + } + if not cols: + return + 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 ────────────────────────────────────────────────────────── def upsert_session( @@ -730,7 +775,12 @@ class ControlPlaneDB: number: int, action: str, ) -> dict[str, Any]: - """Gate a mutation: require active assignment+lease for this session/work.""" + """Gate a mutation: require active assignment+lease for this session/work. + + Fail-closed when the work item is terminal (merged/closed) or when the + assignment's expected_head_sha no longer matches the work item head + (stale-head after assignment time). + """ kind_norm = (kind or "").strip().lower() if kind_norm not in WORK_KINDS: raise InvalidWorkKindError(f"invalid kind '{kind}'") @@ -746,6 +796,12 @@ class ControlPlaneDB: raise LeaseRequiredError( f"no work_item for {kind_norm}#{number}; assign first" ) + work_state = (work["state"] or "").strip().lower() + if work_state in TERMINAL_WORK_STATES: + raise LeaseRequiredError( + f"work item {kind_norm}#{number} is terminal state " + f"'{work_state}'; assignment no longer authorizes mutations" + ) self._expire_stale_leases_conn(conn, work_item_id=int(work["work_item_id"])) asn = conn.execute( """ @@ -769,6 +825,14 @@ 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. + assigned_head = (asn["expected_head_sha"] or "").strip() + current_head = (work["current_head_sha"] or "").strip() + if assigned_head and assigned_head != current_head: + raise LeaseRequiredError( + f"assignment {asn['assignment_id']} stale head: expected " + f"{assigned_head!r} but work item head is {current_head!r}" + ) allowed = json.loads(asn["allowed_actions"]) forbidden = json.loads(asn["forbidden_actions"]) if action in forbidden: @@ -867,6 +931,10 @@ class ControlPlaneDB: status: str = "open", ) -> dict[str, Any]: now_s = _ts() + # Canonical key: never store NULL in UNIQUE scope columns. + base_url = _norm_scope(provider_base_url) + p_org = _norm_scope(provider_org) + p_project = _norm_scope(provider_project) with self._tx() as conn: conn.execute( """ @@ -892,9 +960,9 @@ class ControlPlaneDB: """, ( provider, - provider_base_url, - provider_org, - provider_project, + base_url, + p_org, + p_project, provider_issue_id, provider_short_id, provider_permalink, @@ -914,18 +982,12 @@ class ControlPlaneDB: """ SELECT * FROM incident_links WHERE provider = ? - AND IFNULL(provider_base_url, '') = IFNULL(?, '') - AND IFNULL(provider_org, '') = IFNULL(?, '') - AND IFNULL(provider_project, '') = IFNULL(?, '') + AND provider_base_url = ? + AND provider_org = ? + AND provider_project = ? AND provider_issue_id = ? """, - ( - provider, - provider_base_url, - provider_org, - provider_project, - provider_issue_id, - ), + (provider, base_url, p_org, p_project, provider_issue_id), ).fetchone() return dict(row) diff --git a/docs/architecture/control-plane-db-substrate.md b/docs/architecture/control-plane-db-substrate.md index b531799..5bb589a 100644 --- a/docs/architecture/control-plane-db-substrate.md +++ b/docs/architecture/control-plane-db-substrate.md @@ -1,7 +1,9 @@ # Control-plane DB substrate (#613) -**Status:** Implemented (SQLite single-writer MVP) -**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md) +**Status:** Implemented (SQLite single-writer MVP) + +**ADR:** [`mcp-allocator-control-plane-observability-adr.md`](mcp-allocator-control-plane-observability-adr.md) + **Module:** `control_plane_db.py` ## Architecture statement @@ -14,17 +16,19 @@ |------------|--------| | Schema | `sessions`, `work_items`, `leases`, `assignments`, `terminal_locks`, `events`, `incident_links` | | Atomic assign+lease | `ControlPlaneDB.assign_and_lease` — one `BEGIN IMMEDIATE` transaction | -| Mutation gate | `require_valid_assignment` — workers need a live assignment/lease | +| Mutation gate | `require_valid_assignment` — live lease + allowed action + non-terminal work + non-stale head | | Heartbeat / release / expire | Lease lifecycle helpers | | Terminal-lock index | Routing signal for #600 (terminal path first) | -| `incident_links` | Provider-neutral link model for #612 — **not** assignable work | +| `incident_links` | Provider-neutral link model for #612 — **not** assignable work; scope keys NULL-safe | ## Hard rules (enforced in code) 1. Assignable `work_items.kind` ∈ {`issue`, `pr`} only — **never** raw Sentry/GlitchTip incidents. 2. Two concurrent sessions cannot both receive an active assignment on the same open work item (second gets `wait`). -3. Merged/closed work items return `no_safe_work`. -4. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6). +3. Merged/closed work items return `no_safe_work` at assign time, and `require_valid_assignment` fails closed if the work item becomes terminal later. +4. Assignments pin `expected_head_sha`; mutations fail closed if the work item head drifts. +5. SQLite path is the **single-writer MVP** (`GITEA_CONTROL_PLANE_DB`, default under `~/.cache/gitea-tools/control-plane/`). Multi-session multi-host production requires **Postgres** or a **single allocator daemon** (ADR §6). +6. `incident_links` optional scope fields are stored as empty strings (never NULL) so UNIQUE is canonical across minimal upserts. ## Dependency chain diff --git a/tests/test_control_plane_db.py b/tests/test_control_plane_db.py index 8091272..8535335 100644 --- a/tests/test_control_plane_db.py +++ b/tests/test_control_plane_db.py @@ -36,7 +36,7 @@ class ControlPlaneDBTest(unittest.TestCase): rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall()) finally: conn.close() - self.assertEqual(rows["schema_version"], "1") + self.assertEqual(rows["schema_version"], "2") self.assertIn("DB coordinates", rows["architecture"]) self.assertIn("bridge", rows["architecture"].lower()) @@ -372,6 +372,159 @@ class ControlPlaneDBTest(unittest.TestCase): self.assertEqual(b.outcome, "assigned") self.assertEqual(b.session_id, "s2") + def test_require_valid_assignment_rejects_stale_head(self) -> None: + """Assignment must not authorize mutations after work-item head drifts.""" + self.db.upsert_session(session_id="s1", role="author") + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=42, + expected_head_sha="head-v1", + allowed_actions=("implement",), + ) + # Head drifts after assignment + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=42, + current_head_sha="head-v2", + ) + with self.assertRaises(LeaseRequiredError) as ctx: + self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=42, + action="implement", + ) + self.assertIn("stale head", str(ctx.exception).lower()) + + def test_require_valid_assignment_rejects_terminal_state(self) -> None: + """Assignment must not authorize mutations after work item is merged/closed.""" + self.db.upsert_session(session_id="s1", role="author") + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=55, + expected_head_sha="abc", + allowed_actions=("implement",), + ) + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="pr", + number=55, + state="merged", + current_head_sha="abc", + ) + with self.assertRaises(LeaseRequiredError) as ctx: + self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="pr", + number=55, + action="implement", + ) + self.assertIn("terminal", str(ctx.exception).lower()) + + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + state="open", + ) + self.db.assign_and_lease( + session_id="s1", + role="author", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + allowed_actions=("implement",), + ) + self.db.upsert_work_item( + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + state="closed", + ) + with self.assertRaises(LeaseRequiredError): + self.db.require_valid_assignment( + session_id="s1", + remote="prgs", + org="o", + repo="r", + kind="issue", + number=56, + action="implement", + ) + + 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( + provider="sentry", + provider_issue_id="inc-1", + gitea_org="org", + gitea_repo="repo", + gitea_issue_number=1, + ) + b = self.db.upsert_incident_link( + provider="sentry", + provider_issue_id="inc-1", + gitea_org="org", + gitea_repo="repo", + gitea_issue_number=2, + # omit optional scope fields again + ) + c = self.db.upsert_incident_link( + provider="sentry", + provider_issue_id="inc-1", + gitea_org="org", + gitea_repo="repo", + gitea_issue_number=3, + provider_base_url=None, + provider_org="", + provider_project=" ", + ) + self.assertEqual(a["link_id"], b["link_id"]) + self.assertEqual(b["link_id"], c["link_id"]) + self.assertEqual(c["gitea_issue_number"], 3) + self.assertEqual(c["provider_base_url"], "") + self.assertEqual(c["provider_org"], "") + self.assertEqual(c["provider_project"], "") + + import sqlite3 + + conn = sqlite3.connect(self.db_path) + try: + n = conn.execute( + "SELECT COUNT(*) FROM incident_links WHERE provider_issue_id = ?", + ("inc-1",), + ).fetchone()[0] + finally: + conn.close() + self.assertEqual(n, 1) + if __name__ == "__main__": unittest.main()