fix: safe incident_links migration order; require PR head pin
Address remaining PR #619 blockers on head 783ea88:
- Deduplicate legacy NULL-scope incident_links before normalizing to ''
- Fail closed when duplicate rows disagree on Gitea target
- Require non-empty expected_head_sha for PR assign and mutation
This commit is contained in:
+70
-17
@@ -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 "
|
||||
|
||||
@@ -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(
|
||||
|
||||
Reference in New Issue
Block a user