From 53ce1b1a5ec6476bbe8a153e58f9e13f2bbf1ff9 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sat, 25 Jul 2026 03:32:24 -0400 Subject: [PATCH] fix(webui): compensate stray allocations and keep preview side-effect free MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses both blockers from the PR #902 review (review 589) for issue #643. B1 — an allocator-created assignment could be orphaned and reported as no mutation. apply_request re-previews, then re-runs the allocator with apply=True. The CAS fingerprint hashes only {kind, number} plus exclusions, so a competing lease taken on the requested unit inside the window leaves the fingerprint identical: the pin passes, the selection loop skips the now-claimed unit and commits an assignment on the *next* one, and _selection_matches then fails on egress. The old code returned mutation_performed False with that lease still committed and owned by a synthetic session nothing heartbeats. The window contains a second full load_queue_snapshot(), so it is seconds wide, and foreign sessions acting on this repo concurrently are an observed condition. The egress mismatch now releases the assignment the allocator created before refusing. When the release succeeds the refusal reports mutation_performed False and a compensation record; when it fails the response carries mutation_performed True, an explicit orphaned_assignment, and a gitea_release_workflow_lease reclaim action, because claiming nothing changed while a lease is live is the defect rather than a report of it. The same state was reachable through _run_allocator's bare except Exception: allocate_next_work only catches InvalidWorkKindError, LeaseRequiredError and ControlPlaneError, so anything raised after assign_and_lease committed arrived as "no result" with a durable lease. A None result on the apply path now sweeps and releases whatever this flow's session owns. That sweep is only possible because of the B2 fix below — the session id is now stable across the flow, so the lease is findable. B2 — the "read-only" preview wrote to the control-plane DB. allocate_next_work called db.upsert_session and db.expire_stale_leases unconditionally, before the apply branch was consulted, and default_allocator minted a fresh webui-request- per call. Every preview therefore appended a never-reused session row and mutated global lease state while the payload said dry_run True / mutation_performed False, driven by an operator refreshing a form. One apply wrote two rows and bound the lease to the second, which is why B1's orphan had no reclaimable owner. allocate_next_work gains a keyword-only side_effect_free flag, default False so every existing caller is byte-for-byte unchanged. Under the flag both writes are suppressed and expired leases are instead filtered out of the claim map in memory, which reaches the same selection the sweep would have produced without persisting anything; a claim whose expiry cannot be parsed is kept, since an unreadable expiry is not evidence that work is free. side_effect_free with apply=True fails closed rather than silently reserving. request_service mints one session id per request flow and threads it through both the dry-run and the apply, and the dry-run now routes through the side-effect-free path. Coverage. test_allocator_drift_on_apply_is_not_read_as_an_assignment asserted the defect — it built the orphan state and then required mutation_performed to be False, which a leak satisfies. It now requires the compensating release. Added: release-failure surfacing a reclaim action, an assignment with no lease id, the post-commit exception route, and session-id identity across the flow. The two areas the review named as having zero coverage now have it: default_allocator past its two fail-closed early returns (side_effect_free routing, session-id pass-through and minting, scope and fingerprint propagation) and default_claims_source (scoped read, and an unreadable substrate denying rather than reading as "nothing claimed"). allocator_service gains side-effect-free tests against a real temp-file DB plus the expiry-filter unit tests. Every new guard was mutation-tested by reverting it one at a time: B1 compensation removed → 3 failures; post-commit sweep removed → 1; session id re-minted per call → 2; side_effect_free ignored → 3; in-memory expiry filter removed → 1. Verification: WEBUI_TEST_OFFLINE=1 python -m pytest tests/ -q from this branches/ worktree gives 23 failed, 5263 passed, 6 skipped, 899 subtests. The sorted FAILED set is identical to the reviewer's clean-master baseline at 2f4dec83 (23 failed, 5190 passed) — no new, changed, or disappeared failure — and 21 tests were added over the reviewed head's 5242. Zero conflict markers; py_compile passes; git diff --check clean. Refs #643, PR #902 Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01V6xFqovhbArPv61j9KCGkL --- allocator_service.py | 122 +++++++-- tests/test_allocator_service.py | 158 ++++++++++++ tests/test_webui_request_initiation.py | 327 ++++++++++++++++++++++++- webui/request_service.py | 282 +++++++++++++++++++-- 4 files changed, 839 insertions(+), 50 deletions(-) diff --git a/allocator_service.py b/allocator_service.py index 9f32fd9..a085d25 100644 --- a/allocator_service.py +++ b/allocator_service.py @@ -23,6 +23,7 @@ import json import os import uuid from dataclasses import dataclass, field +from datetime import datetime, timezone from typing import Any, Mapping, Sequence from control_plane_db import ( @@ -738,6 +739,46 @@ def normalize_exclude_issue_numbers( return sorted(out) +def _claim_expires_at(claim: Any) -> datetime | None: + """Parse a claim's ``expires_at``, or ``None`` when it is absent/malformed.""" + if not isinstance(claim, Mapping): + return None + text = str(claim.get("expires_at") or "").strip() + if not text: + return None + if text.endswith("Z"): + text = text[:-1] + "+00:00" + try: + parsed = datetime.fromisoformat(text) + except ValueError: + return None + if parsed.tzinfo is None: + parsed = parsed.replace(tzinfo=timezone.utc) + return parsed.astimezone(timezone.utc) + + +def _drop_expired_claims( + claims: Mapping[tuple[str, int], dict[str, Any]], + *, + now: datetime | None = None, +) -> dict[tuple[str, int], dict[str, Any]]: + """Claims minus those whose lease has already expired (#643). + + The read-only mirror of ``expire_stale_leases``: the sweep marks such rows + ``expired`` so they stop being returned as claims, and this reaches the same + view without writing. A claim with no parseable ``expires_at`` is **kept** — + an unreadable expiry is not evidence that work is free. + """ + moment = now or datetime.now(timezone.utc) + kept: dict[tuple[str, int], dict[str, Any]] = {} + for key, claim in (claims or {}).items(): + expires_at = _claim_expires_at(claim) + if expires_at is not None and expires_at <= moment: + continue + kept[key] = claim + return kept + + def candidate_set_fingerprint( candidates: Sequence[WorkCandidate], *, @@ -826,12 +867,22 @@ def allocate_next_work( exclude_issue_numbers: Sequence[int] | None = None, expected_candidate_set_fingerprint: str | None = None, allocation_mode: str | None = None, + side_effect_free: bool = False, ) -> dict[str, Any]: """Select and optionally reserve the next work unit via control-plane DB. *apply=False* (default): dry-run selection only — no lease/assignment. *apply=True*: atomic ``assign_and_lease`` for the selected candidate. + *side_effect_free* (#643): a dry run that writes **nothing** to the + control-plane DB. A plain ``apply=False`` still registered a session row and + swept stale leases globally, so a caller advertising a read-only preview was + mutating on every call. Under this flag both writes are suppressed and stale + leases are instead filtered out of the claim map in memory, which yields the + same selection the sweep would have produced without persisting anything. + Incompatible with *apply* — the combination fails closed rather than + silently reserving. + *allocation_mode* (#840): ``cross_role`` (default for controller) inspects the complete queue and returns one authoritative selection naming the required downstream role/profile/action. ``role_scoped`` keeps prior @@ -885,40 +936,57 @@ def allocate_next_work( "allocation_mode": (allocation_mode or "").strip() or None, } - session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}" - try: - db.upsert_session( - session_id=session_id, - role=role_norm, - profile=profile_name, - pid=os.getpid(), - controller_instance_id=controller_instance_id, - ) - except Exception as exc: # noqa: BLE001 — surface structured + # A side-effect-free run may never reserve: reserving is a write, and the + # flag is the caller's assertion that this call writes nothing (#643). + if side_effect_free and apply: return { "success": False, "outcome": OUTCOME_NO_SAFE, + "apply": True, "reasons": [ - f"failed to register session in control-plane DB: {exc} " - "(fail closed, #613)" + "side_effect_free is incompatible with apply=True; an " + "assignment is a write (fail closed, #643)" ], "skipped": [], "assignment": None, "substrate": "control_plane_db", } - # Expire stale leases globally before selection. - try: - db.expire_stale_leases() - except Exception as exc: # noqa: BLE001 - return { - "success": False, - "outcome": OUTCOME_NO_SAFE, - "reasons": [f"lease expiry failed: {exc} (fail closed)"], - "skipped": [], - "assignment": None, - "substrate": "control_plane_db", - } + session_id = (session_id or "").strip() or f"alloc-{uuid.uuid4().hex[:12]}" + if not side_effect_free: + try: + db.upsert_session( + session_id=session_id, + role=role_norm, + profile=profile_name, + pid=os.getpid(), + controller_instance_id=controller_instance_id, + ) + except Exception as exc: # noqa: BLE001 — surface structured + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "reasons": [ + f"failed to register session in control-plane DB: {exc} " + "(fail closed, #613)" + ], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } + + # Expire stale leases globally before selection. + try: + db.expire_stale_leases() + except Exception as exc: # noqa: BLE001 + return { + "success": False, + "outcome": OUTCOME_NO_SAFE, + "reasons": [f"lease expiry failed: {exc} (fail closed)"], + "skipped": [], + "assignment": None, + "substrate": "control_plane_db", + } terminal = None try: @@ -953,6 +1021,12 @@ def allocate_next_work( "assignment": None, "substrate": "control_plane_db", } + if side_effect_free: + # ``list_active_claims`` filters on status alone, so without the + # global sweep an already-expired lease would still read as a live + # claim and the preview would report work as taken that is free. + # Drop those in memory: same view the sweep produces, no write. + claims = _drop_expired_claims(claims) try: exclude_nums = normalize_exclude_issue_numbers(exclude_issue_numbers) diff --git a/tests/test_allocator_service.py b/tests/test_allocator_service.py index 3e252a1..5c37d0d 100644 --- a/tests/test_allocator_service.py +++ b/tests/test_allocator_service.py @@ -7,6 +7,7 @@ import tempfile import threading import unittest from concurrent.futures import ThreadPoolExecutor, as_completed +from datetime import datetime, timezone from allocator_service import ( OUTCOME_ASSIGNED, @@ -15,6 +16,7 @@ from allocator_service import ( OUTCOME_PREVIEW, OUTCOME_WAIT, WorkCandidate, + _drop_expired_claims, allocate_next_work, candidate_from_dict, classify_skip, @@ -362,5 +364,161 @@ class AllocatorServiceTest(unittest.TestCase): self.assertIn("unavailable", res["reasons"][0].lower()) +class SideEffectFreeAllocationTest(unittest.TestCase): + """``side_effect_free`` dry runs write nothing to the control plane (#643). + + A plain ``apply=False`` still called ``upsert_session`` and + ``expire_stale_leases`` before the apply branch was consulted, so a caller + advertising a read-only preview mutated on every call — one unreferenced + session row per preview, plus a global lease sweep. + """ + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.db = ControlPlaneDB(os.path.join(self._tmp.name, "cp.sqlite3")) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def _alloc(self, **kwargs): + defaults = dict( + db=self.db, + session_id="s-preview", + role="author", + remote="prgs", + org="org", + repo="repo", + candidates=[ + WorkCandidate(kind="issue", number=643, labels=("status:ready",)) + ], + apply=False, + profile_name="prgs-author", + username="jcwalker3", + ) + defaults.update(kwargs) + return allocate_next_work(**defaults) + + def _session_ids(self) -> set[str]: + return {str(r.get("session_id")) for r in self.db.list_sessions()} + + def test_side_effect_free_preview_writes_no_session_row(self): + before = self._session_ids() + result = self._alloc(side_effect_free=True) + self.assertEqual(result["outcome"], OUTCOME_PREVIEW) + self.assertEqual(self._session_ids(), before) + self.assertNotIn("s-preview", self._session_ids()) + + def test_plain_dry_run_still_registers_a_session(self): + # The default is unchanged for every existing caller. + self._alloc() + self.assertIn("s-preview", self._session_ids()) + + def test_repeated_previews_do_not_accumulate_rows(self): + for index in range(5): + self._alloc(side_effect_free=True, session_id=f"s-{index}") + self.assertEqual(self._session_ids(), set()) + + def test_side_effect_free_does_not_sweep_stale_leases(self): + self.db.upsert_session(session_id="owner", role="author", pid=1) + assigned = self.db.assign_and_lease( + session_id="owner", + role="author", + remote="prgs", + org="org", + repo="repo", + kind="issue", + number=999, + lease_ttl_seconds=-60, # already expired + ) + self.assertEqual(assigned.outcome, "assigned") + + self._alloc(side_effect_free=True) + + # The expired row is still 'active' in the DB: nothing swept it. + statuses = { + r["lease_id"]: r["status"] + for r in self.db.list_leases( + remote="prgs", org="org", repo="repo", + statuses=("active", "expired"), + ) + } + self.assertEqual(statuses.get(assigned.lease_id), "active") + + def test_expired_claims_are_filtered_in_memory_so_work_stays_selectable(self): + """The read-only mirror of the sweep: expired claims must not block.""" + self.db.upsert_session(session_id="owner", role="author", pid=1) + self.db.assign_and_lease( + session_id="owner", + role="author", + remote="prgs", + org="org", + repo="repo", + kind="issue", + number=643, + lease_ttl_seconds=-60, # expired: must not withhold #643 + ) + result = self._alloc(side_effect_free=True) + self.assertEqual(result["outcome"], OUTCOME_PREVIEW) + self.assertEqual(result["selected"]["number"], 643) + + def test_a_live_claim_still_withholds_the_work(self): + self.db.upsert_session(session_id="owner", role="author", pid=1) + self.db.assign_and_lease( + session_id="owner", + role="author", + remote="prgs", + org="org", + repo="repo", + kind="issue", + number=643, + lease_ttl_seconds=3600, + ) + result = self._alloc(side_effect_free=True) + self.assertNotEqual(result["outcome"], OUTCOME_ASSIGNED) + self.assertNotEqual((result.get("selected") or {}).get("number"), 643) + + def test_side_effect_free_with_apply_fails_closed(self): + result = self._alloc(side_effect_free=True, apply=True) + self.assertFalse(result["success"]) + self.assertEqual(result["outcome"], OUTCOME_NO_SAFE) + self.assertIsNone(result["assignment"]) + self.assertIn("incompatible with apply", result["reasons"][0]) + # And it reserved nothing. + self.assertEqual( + self.db.list_leases(remote="prgs", org="org", repo="repo"), [] + ) + + +class DropExpiredClaimsTest(unittest.TestCase): + """The in-memory expiry filter behind side-effect-free previews (#643).""" + + def test_unparseable_expiry_is_kept_rather_than_assumed_free(self): + claims = { + ("issue", 1): {"lease_id": "l1", "expires_at": "not-a-date"}, + ("issue", 2): {"lease_id": "l2"}, + ("issue", 3): {"lease_id": "l3", "expires_at": None}, + } + self.assertEqual(_drop_expired_claims(claims), claims) + + def test_expired_dropped_and_future_kept(self): + now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc) + claims = { + ("issue", 1): {"expires_at": "2026-07-25T11:59:59+00:00"}, + ("issue", 2): {"expires_at": "2026-07-25T12:00:01+00:00"}, + ("issue", 3): {"expires_at": "2026-07-25T12:00:00+00:00"}, # boundary + } + kept = _drop_expired_claims(claims, now=now) + self.assertEqual(set(kept), {("issue", 2)}) + + def test_naive_and_zulu_timestamps_are_treated_as_utc(self): + now = datetime(2026, 7, 25, 12, 0, tzinfo=timezone.utc) + claims = { + ("issue", 1): {"expires_at": "2026-07-25T11:00:00"}, # naive, past + ("issue", 2): {"expires_at": "2026-07-25T13:00:00Z"}, # zulu, future + } + kept = _drop_expired_claims(claims, now=now) + self.assertEqual(set(kept), {("issue", 2)}) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_webui_request_initiation.py b/tests/test_webui_request_initiation.py index c733898..e05978c 100644 --- a/tests/test_webui_request_initiation.py +++ b/tests/test_webui_request_initiation.py @@ -16,6 +16,7 @@ allocator refuses an incomplete inventory rather than ranking a partial set. from __future__ import annotations +import contextlib import json import os import pathlib @@ -89,6 +90,63 @@ def _selection( } +@contextlib.contextmanager +def _patched_control_plane( + *, + release_sink: list[tuple[str, str]] | None = None, + release_error: Exception | None = None, + leases=None, +): + """Stand in for ``control_plane_db`` so compensation paths are observable. + + The service imports the module inside the function, so patching + ``sys.modules`` is what intercepts it. No real DB is opened. + """ + module = mock.MagicMock() + db = mock.MagicMock() + + def _release(lease_id, *, session_id): + if release_error is not None: + raise release_error + if release_sink is not None: + release_sink.append((lease_id, session_id)) + + db.release_lease.side_effect = _release + db.list_leases.side_effect = leases or (lambda **_kwargs: []) + module.ControlPlaneDB.return_value = db + with mock.patch.dict(sys.modules, {"control_plane_db": module}): + yield db + + +def _drifting_allocator(*, lease_id: str | None = "lease-wrong"): + """An allocator that previews the requested unit but assigns another. + + This is the #643 B1 race in miniature: the CAS fingerprint hashes only + ``{kind, number}``, so a lease taken on the requested unit inside the window + leaves the fingerprint identical while the selection moves on. + """ + assignment: dict[str, Any] = {"assignment_id": "asn-wrong"} + if lease_id: + assignment["lease_id"] = lease_id + + def _drifting( + *, request, apply, expected_candidate_set_fingerprint=None, session_id=None + ): + return { + "outcome": ( + allocator_service.OUTCOME_ASSIGNED + if apply + else allocator_service.OUTCOME_PREVIEW + ), + "selected": _selection(number=999 if apply else 643), + "assignment": dict(assignment) if apply else None, + "candidate_set_fingerprint": "fp-test", + "session_id": session_id, + } + + return _drifting + + def _fake_allocator( *, selection: dict[str, Any] | None = None, @@ -587,30 +645,170 @@ class TestApplyOutcomes(unittest.TestCase): self.assertFalse(result["mutation_performed"]) def test_allocator_drift_on_apply_is_not_read_as_an_assignment(self): - """The apply call must return *this* work unit, not a substitute.""" + """The apply call must return *this* work unit, not a substitute. - def _drifting(*, request, apply, expected_candidate_set_fingerprint=None): + Drift is not merely refused: the allocator has already committed the + substitute assignment by the time egress rejects it, so the refusal must + also release it. Asserting only ``mutation_performed is False`` would + pass just as well against a leak. + """ + released: list[tuple[str, str]] = [] + + with _patched_control_plane(release_sink=released): + result = request_service.apply_request( + _request(number=643), + principal=_principal(console_authz.OPERATOR), + confirm=True, + allocator=_drifting_allocator(), + claims_source=_no_claims, + ) + + self.assertFalse(result["ok"]) + self.assertIsNone(result["assignment"]) + # The substitute assignment was released, so nothing durable survives. + self.assertEqual(len(released), 1) + self.assertEqual(released[0][0], "lease-wrong") + compensation = result["compensation"] + self.assertTrue(compensation["released"]) + self.assertEqual(compensation["lease_id"], "lease-wrong") + self.assertEqual(compensation["assignment_id"], "asn-wrong") + self.assertEqual(compensation["selected"]["number"], 999) + self.assertFalse(result["mutation_performed"]) + self.assertNotIn("orphaned_assignment", result) + + def test_drift_whose_release_fails_reports_the_orphan_and_a_reclaim(self): + """A release that fails must surface the leak, never swallow it.""" + with _patched_control_plane(release_error=RuntimeError("db is read-only")): + result = request_service.apply_request( + _request(number=643), + principal=_principal(console_authz.OPERATOR), + confirm=True, + allocator=_drifting_allocator(), + claims_source=_no_claims, + ) + + self.assertFalse(result["ok"]) + self.assertIsNone(result["assignment"]) + # A lease really is out there; saying "nothing changed" would be a lie. + self.assertTrue(result["mutation_performed"]) + orphan = result["orphaned_assignment"] + self.assertFalse(orphan["released"]) + self.assertTrue(orphan["attempted"]) + self.assertIn("db is read-only", orphan["error"]) + reclaim = orphan["reclaim_action"] + self.assertEqual(reclaim["tool"], "gitea_release_workflow_lease") + self.assertEqual(reclaim["lease_id"], "lease-wrong") + + def test_drift_without_a_lease_id_still_surfaces_a_reclaim(self): + """An assignment with no lease id cannot be released — say so.""" + result = request_service.apply_request( + _request(number=643), + principal=_principal(console_authz.OPERATOR), + confirm=True, + allocator=_drifting_allocator(lease_id=None), + claims_source=_no_claims, + ) + self.assertFalse(result["ok"]) + self.assertTrue(result["mutation_performed"]) + orphan = result["orphaned_assignment"] + self.assertFalse(orphan["attempted"]) + self.assertEqual(orphan["assignment_id"], "asn-wrong") + self.assertIn("reclaim_action", orphan) + + def test_exception_after_commit_releases_the_session_lease(self): + """A post-commit exception surfaces as no result — with a live lease. + + ``allocate_next_work`` catches only three exception types, so anything + else raised after ``assign_and_lease`` committed reaches the caller as + ``None`` while the lease is durable. The stable per-flow session id is + what makes that lease findable. + """ + released: list[tuple[str, str]] = [] + seen: list[str | None] = [] + + def _explodes_after_commit( + *, request, apply, expected_candidate_set_fingerprint=None, session_id=None + ): + seen.append(session_id) + if not apply: + return { + "outcome": allocator_service.OUTCOME_PREVIEW, + "selected": _selection(number=643), + "assignment": None, + "candidate_set_fingerprint": "fp-test", + } + raise KeyError("selection['required_profile']") + + def _leases(**_kwargs): + return [ + { + "lease_id": "lease-committed", + "session_id": seen[-1], + "work_kind": "issue", + "work_number": 643, + } + ] + + with _patched_control_plane(release_sink=released, leases=_leases): + result = request_service.apply_request( + _request(number=643), + principal=_principal(console_authz.OPERATOR), + confirm=True, + allocator=_explodes_after_commit, + claims_source=_no_claims, + ) + + self.assertFalse(result["ok"]) + self.assertEqual( + result["reason_code"], request_service.REASON_EVIDENCE_UNAVAILABLE + ) + self.assertEqual(len(released), 1) + self.assertEqual(released[0][0], "lease-committed") + self.assertEqual( + result["compensation"]["released"][0]["lease_id"], "lease-committed" + ) + self.assertFalse(result["mutation_performed"]) + + def test_one_session_id_spans_the_dry_run_and_the_apply(self): + """Both halves of an apply share one control-plane identity.""" + seen: list[str | None] = [] + + def _recording( + *, request, apply, expected_candidate_set_fingerprint=None, session_id=None + ): + seen.append(session_id) return { "outcome": ( allocator_service.OUTCOME_ASSIGNED if apply else allocator_service.OUTCOME_PREVIEW ), - "selected": _selection(number=999 if apply else 643), - "assignment": {"assignment_id": "asn-wrong"} if apply else None, + "selected": _selection(number=643), + "assignment": ( + { + "assignment_id": "asn-ok", + "lease_id": "lease-ok", + "expected_head_sha": None, + } + if apply + else None + ), "candidate_set_fingerprint": "fp-test", + "session_id": session_id, } result = request_service.apply_request( _request(number=643), principal=_principal(console_authz.OPERATOR), confirm=True, - allocator=_drifting, + allocator=_recording, claims_source=_no_claims, ) - self.assertFalse(result["ok"]) - self.assertIsNone(result["assignment"]) - self.assertFalse(result["mutation_performed"]) + self.assertTrue(result["ok"]) + self.assertEqual(len(seen), 2) + self.assertTrue(all(s for s in seen)) + self.assertEqual(seen[0], seen[1], "dry-run and apply must share one id") + self.assertTrue(seen[0].startswith("webui-request-")) def test_viewer_cannot_apply(self): result = request_service.apply_request( @@ -634,6 +832,119 @@ class TestApplyOutcomes(unittest.TestCase): self.assertFalse(result["mutation_performed"]) +class TestDefaultAllocatorPastTheEarlyReturns(unittest.TestCase): + """``default_allocator`` beyond its two fail-closed guards (#643). + + Both prior tests returned before ``ControlPlaneDB`` was ever constructed, so + the session id, the ``side_effect_free`` routing and the CAS round-trip had + no coverage at all — which is how a preview that writes session rows shipped. + """ + + def setUp(self): + from webui.queue_loader import QueueSnapshot + + self.snapshot = QueueSnapshot( + project_id="p", + repo_label="r", + prs=(), + issues=(), + pr_pagination=None, + issue_pagination=None, + ) + + @contextlib.contextmanager + def _harness(self): + """Run the real ``default_allocator`` against a recorded allocator call.""" + calls: dict[str, Any] = {} + + def _allocate(db, **kwargs): + calls.update(kwargs) + return {"outcome": allocator_service.OUTCOME_PREVIEW} + + module = mock.MagicMock() + with mock.patch.dict(sys.modules, {"control_plane_db": module}), \ + mock.patch( + "webui.queue_loader.load_queue_snapshot", + return_value=self.snapshot, + ), \ + mock.patch( + "webui.traffic_loader.candidates_from_queue_snapshot", + return_value=[], + ), \ + mock.patch.object( + allocator_service, "allocate_next_work", side_effect=_allocate + ): + yield calls + + def test_preview_runs_side_effect_free_and_never_applies(self): + with self._harness() as calls: + result = request_service.default_allocator( + request=_request(), apply=False + ) + self.assertIsNotNone(result) + self.assertTrue(calls["side_effect_free"]) + self.assertFalse(calls["apply"]) + + def test_apply_is_not_side_effect_free(self): + with self._harness() as calls: + request_service.default_allocator(request=_request(), apply=True) + self.assertFalse(calls["side_effect_free"]) + self.assertTrue(calls["apply"]) + + def test_caller_session_id_is_passed_through_verbatim(self): + with self._harness() as calls: + request_service.default_allocator( + request=_request(), apply=True, session_id="webui-request-fixed" + ) + self.assertEqual(calls["session_id"], "webui-request-fixed") + + def test_absent_session_id_is_minted_with_the_expected_shape(self): + with self._harness() as calls: + request_service.default_allocator(request=_request(), apply=False) + self.assertTrue(str(calls["session_id"]).startswith("webui-request-")) + + def test_scope_and_fingerprint_reach_the_allocator(self): + with self._harness() as calls: + request_service.default_allocator( + request=_request(number=664), + apply=False, + expected_candidate_set_fingerprint="fp-pinned", + ) + self.assertEqual(calls["expected_candidate_set_fingerprint"], "fp-pinned") + self.assertEqual(calls["remote"], SCOPE["remote"]) + self.assertEqual(calls["org"], SCOPE["org"]) + self.assertEqual(calls["repo"], SCOPE["repo"]) + self.assertEqual(calls["allocation_mode"], "role_scoped") + + +class TestDefaultClaimsSource(unittest.TestCase): + """``default_claims_source`` had no test at all (#643).""" + + def test_claims_are_read_for_the_request_scope(self): + db = mock.MagicMock() + db.list_active_claims.return_value = {("issue", 643): {"lease_id": "l1"}} + module = mock.MagicMock() + module.ControlPlaneDB.return_value = db + with mock.patch.dict(sys.modules, {"control_plane_db": module}): + claims = request_service.default_claims_source(_request()) + self.assertEqual(claims, {("issue", 643): {"lease_id": "l1"}}) + db.list_active_claims.assert_called_once_with( + remote=SCOPE["remote"], org=SCOPE["org"], repo=SCOPE["repo"] + ) + + def test_an_unreadable_substrate_denies_rather_than_returning_empty(self): + module = mock.MagicMock() + module.ControlPlaneDB.side_effect = RuntimeError("no db") + with mock.patch.dict(sys.modules, {"control_plane_db": module}): + # _load_claims converts the failure into None, which fails the + # lease check closed; an empty mapping would read as "nothing + # claimed" and wrongly authorize. + claims = request_service._load_claims( + _request(), request_service.default_claims_source + ) + self.assertIsNone(claims) + + class TestAllocatorIntegrationFakes(unittest.TestCase): """The real default allocator refuses a partial inventory (#758).""" diff --git a/webui/request_service.py b/webui/request_service.py index 6941027..5926971 100644 --- a/webui/request_service.py +++ b/webui/request_service.py @@ -35,6 +35,7 @@ the resulting assignment by ``correlation_id``. from __future__ import annotations +import inspect import uuid from dataclasses import dataclass, field from typing import Any, Callable, Mapping, Sequence @@ -321,6 +322,17 @@ def _correlation_id() -> str: return f"req-{uuid.uuid4().hex}" +def _new_session_id() -> str: + """Mint the control-plane session id for one request flow (#643). + + Minted once per flow and reused by the dry-run and the apply, so a lease the + apply creates is owned by an id the caller still holds and can release. When + each call minted its own id, an apply wrote two session rows and bound the + lease to the second — leaving it with no owner able to reclaim it. + """ + return f"webui-request-{uuid.uuid4().hex[:12]}" + + def _selection_matches( selection: Mapping[str, Any] | None, request: WorkRequest ) -> bool: @@ -561,8 +573,15 @@ def preview_request( claims_source: ClaimsFn | None = None, correlation_id: str | None = None, audit: bool = True, + session_id: str | None = None, ) -> RequestPreview: - """Build the read-only intent preview for *request*. Never mutates.""" + """Build the read-only intent preview for *request*. Never mutates. + + "Never mutates" is now literal on the default path: the allocator runs + ``side_effect_free``, so no control-plane session row is written and no lease + sweep runs. *session_id* is threaded from an enclosing apply so both halves + of that flow share one control-plane identity (#643). + """ who = principal or console_authz.ANONYMOUS corr = correlation_id or _correlation_id() decision = console_authz.authorize(ACTION_ID, who, for_execution=False) @@ -573,7 +592,9 @@ def preview_request( if decision.allowed: # An unauthorized principal never reaches the allocator or the # control-plane DB: a denial must not double as a queue oracle. - allocation = _run_allocator(request, allocator, apply=False) + allocation = _run_allocator( + request, allocator, apply=False, session_id=session_id + ) claims = _load_claims(request, claims_source) checks.append(_capability_check(request)) checks.append(_lease_check(request, claims)) @@ -650,9 +671,20 @@ def apply_request( The only path to an assignment is the allocator agreeing, on a dry-run, that this work unit is what the requested role should take next. Every refusal returns before any mutation is attempted. + + One exception is unavoidable and is compensated rather than prevented: the + allocator can commit an assignment and only then reveal, on egress, that it + selected a different work unit than the one requested. That assignment is + released before refusing, and ``mutation_performed`` reports whether any + durable control-plane state survives this call — ``False`` once the stray + assignment is gone, ``True`` with an explicit ``orphaned_assignment`` and + reclaim action when the release did not succeed (#643). """ who = principal or console_authz.ANONYMOUS corr = correlation_id or _correlation_id() + # One identity for the whole flow, so a lease the apply creates is owned by + # an id this function still holds and can release. + session_id = _new_session_id() execution_decision = console_authz.authorize(ACTION_ID, who, for_execution=True) authorization = execution_decision.to_dict() @@ -664,6 +696,7 @@ def apply_request( *, status: int, extra: dict[str, Any] | None = None, + mutation_performed: bool = False, ) -> dict[str, Any]: _audit( request, @@ -689,7 +722,9 @@ def apply_request( "authorization": authorization, "assignment": None, "correlation_id": corr, - "mutation_performed": False, + # True only when durable control-plane state survives this refusal — + # an assignment the allocator created that could not be released. + "mutation_performed": bool(mutation_performed), "status_code": status, } payload.update(extra or {}) @@ -724,6 +759,7 @@ def apply_request( claims_source=claims_source, correlation_id=corr, audit=False, + session_id=session_id, ) if not preview.authorized: outcome = ( @@ -747,21 +783,37 @@ def apply_request( allocator, apply=True, expected_candidate_set_fingerprint=fingerprint, + session_id=session_id, ) if not allocation: + # "No result" is not proof of "no write": allocate_next_work only catches + # three exception types, so anything raised after assign_and_lease + # committed lands here with a lease already durable. Release whatever + # this flow's session owns before refusing. + sweep = _sweep_session_leases(request, session_id=session_id) + extra: dict[str, Any] = {} + detail = "the allocator returned no result; nothing was assigned." + if sweep.get("released"): + extra["compensation"] = sweep + detail = ( + "the allocator returned no result after creating an assignment; " + "the assignment was released and nothing remains assigned." + ) + elif sweep.get("error"): + extra["compensation"] = sweep return _refuse( OUTCOME_WAIT, REASON_EVIDENCE_UNAVAILABLE, - "the allocator returned no result; nothing was assigned.", + detail, status=503, + extra=extra or None, ) assignment = allocation.get("assignment") or None outcome = _clean(allocation.get("outcome")) + committed = bool(outcome == allocator_service.OUTCOME_ASSIGNED and assignment) assigned = bool( - outcome == allocator_service.OUTCOME_ASSIGNED - and assignment - and _selection_matches(allocation.get("selected"), request) + committed and _selection_matches(allocation.get("selected"), request) ) if not assigned: blocked = outcome in { @@ -769,15 +821,45 @@ def apply_request( allocator_service.OUTCOME_BLOCKED_TERMINAL, allocator_service.OUTCOME_BLOCKED_EXCLUDED_OWN_LEASE, } + extra = {"allocator_evidence": _allocator_evidence(allocation)} + detail = ( + f"the allocator returned {outcome or 'no outcome'} rather than " + "an assignment for this work unit; nothing was assigned." + ) + if committed: + # The allocator assigned a *different* unit than the one requested. + # That assignment is real and durable; releasing it is the whole + # difference between a refusal and a silent leak. + compensation = _release_stray_assignment( + request, allocation, session_id=session_id + ) + extra["compensation"] = compensation + if compensation.get("released"): + detail = ( + "the allocator assigned a different work unit than the one " + "requested; that assignment was released and nothing " + "remains assigned." + ) + else: + extra["orphaned_assignment"] = compensation + return _refuse( + OUTCOME_BLOCKED if blocked else OUTCOME_WAIT, + REASON_ALLOCATOR_OUTCOME, + ( + "the allocator assigned a different work unit than the " + "one requested and it could not be released; it must be " + "reclaimed explicitly." + ), + status=409, + extra=extra, + mutation_performed=True, + ) return _refuse( OUTCOME_BLOCKED if blocked else OUTCOME_WAIT, REASON_ALLOCATOR_OUTCOME, - ( - f"the allocator returned {outcome or 'no outcome'} rather than " - "an assignment for this work unit; nothing was assigned." - ), + detail, status=409, - extra={"allocator_evidence": _allocator_evidence(allocation)}, + extra=extra, ) _audit( @@ -853,19 +935,175 @@ def _run_allocator( *, apply: bool, expected_candidate_set_fingerprint: str | None = None, + session_id: str | None = None, ) -> dict[str, Any] | None: fn = allocator or default_allocator + kwargs: dict[str, Any] = { + "request": request, + "apply": apply, + "expected_candidate_set_fingerprint": expected_candidate_set_fingerprint, + } + if session_id: + # Injected allocators in tests predate this argument; only pass it to + # callables that accept it so a fake signature is never broken. + if _accepts_session_id(fn): + kwargs["session_id"] = session_id try: - result = fn( - request=request, - apply=apply, - expected_candidate_set_fingerprint=expected_candidate_set_fingerprint, - ) + result = fn(**kwargs) except Exception: # noqa: BLE001 — an allocator failure denies, never proceeds return None return result if isinstance(result, dict) else None +def _accepts_session_id(fn: AllocatorFn) -> bool: + try: + params = inspect.signature(fn).parameters + except (TypeError, ValueError): # builtins / C callables + return False + if "session_id" in params: + return True + return any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()) + + +def _release_stray_assignment( + request: WorkRequest, + allocation: Mapping[str, Any], + *, + session_id: str | None, +) -> dict[str, Any]: + """Release an assignment the allocator committed for the wrong work unit. + + The apply path can commit an assignment and then discover, on egress, that + the allocator selected a *different* unit than the one requested — the CAS + fingerprint hashes only ``{kind, number}`` plus exclusions, so a lease taken + on the requested unit inside the window leaves the fingerprint identical and + the pin passes while the selection moves on. Without compensation that lease + is orphaned: owned by a session nothing heartbeats, and invisible because the + response says nothing was mutated. + + Returns a record describing what was attempted so the caller can report it, + including a reclaim action when the release itself did not succeed. + """ + assignment = allocation.get("assignment") or {} + lease_id = _clean(assignment.get("lease_id")) or None + assignment_id = _clean(assignment.get("assignment_id")) or None + owner = _clean(allocation.get("session_id")) or session_id or None + selected = allocation.get("selected") or {} + record: dict[str, Any] = { + "attempted": False, + "released": False, + "lease_id": lease_id, + "assignment_id": assignment_id, + "owner_session_id": owner, + "selected": { + "kind": _clean(selected.get("kind")).lower() or None, + "number": selected.get("number"), + }, + } + if not lease_id: + # Nothing durable to release (or the allocator reported no lease id); + # still surface the assignment id so a leak is never silent. + record["detail"] = ( + "the allocator reported an assignment with no lease id; nothing " + "could be released automatically." + ) + if assignment_id: + record["reclaim_action"] = _reclaim_action(record) + return record + if not owner: + record["detail"] = ( + "the owning session id is unknown; the assignment cannot be " + "released automatically." + ) + record["reclaim_action"] = _reclaim_action(record) + return record + + record["attempted"] = True + try: + import control_plane_db + + control_plane_db.ControlPlaneDB().release_lease(lease_id, session_id=owner) + except Exception as exc: # noqa: BLE001 — a failed release must be reported + record["error"] = str(exc) + record["detail"] = ( + "the allocator created an assignment for a different work unit and " + "releasing it failed; it must be reclaimed explicitly." + ) + record["reclaim_action"] = _reclaim_action(record) + return record + + record["released"] = True + record["detail"] = ( + "the allocator created an assignment for a different work unit; it was " + "released, so no assignment persists from this request." + ) + return record + + +def _reclaim_action(record: Mapping[str, Any]) -> dict[str, Any]: + """The explicit operator action for an assignment that outlived its request.""" + lease_id = record.get("lease_id") + return { + "tool": "gitea_release_workflow_lease", + "lease_id": lease_id, + "session_id": record.get("owner_session_id"), + "assignment_id": record.get("assignment_id"), + "instructions": ( + "A control-plane assignment was created for a work unit other than " + "the requested one and could not be released automatically. Call " + f"gitea_release_workflow_lease(lease_id={lease_id!r}, " + f"session_id={record.get('owner_session_id')!r}) to reclaim it, or " + "wait for the lease TTL to expire." + ), + } + + +def _sweep_session_leases( + request: WorkRequest, *, session_id: str | None +) -> dict[str, Any]: + """Release any lease this flow's session owns after an indeterminate apply. + + ``_run_allocator`` returns ``None`` for *any* exception, and + ``allocate_next_work`` only catches ``InvalidWorkKindError``, + ``LeaseRequiredError`` and ``ControlPlaneError`` — so an unexpected error + raised *after* ``assign_and_lease`` committed surfaces as "no result" with a + lease already written. Because the session id is now stable across the flow, + that lease is findable: anything this session owns after a failed apply is by + definition unclaimed by anyone, so it is released. + """ + record: dict[str, Any] = {"attempted": False, "released": [], "session_id": session_id} + if not session_id: + return record + record["attempted"] = True + try: + import control_plane_db + + db = control_plane_db.ControlPlaneDB() + leases = db.list_leases( + remote=request.remote, + org=request.org, + repo=request.repo, + statuses=("active",), + ) + for row in leases: + if _clean(row.get("session_id")) != session_id: + continue + lease_id = _clean(row.get("lease_id")) or None + if not lease_id: + continue + db.release_lease(lease_id, session_id=session_id) + record["released"].append( + { + "lease_id": lease_id, + "work_kind": row.get("work_kind"), + "work_number": row.get("work_number"), + } + ) + except Exception as exc: # noqa: BLE001 — best-effort; never mask the refusal + record["error"] = str(exc) + return record + + def _load_claims( request: WorkRequest, claims_source: ClaimsFn | None ) -> Mapping[tuple[str, int], dict[str, Any]] | None: @@ -894,12 +1132,19 @@ def default_allocator( request: WorkRequest, apply: bool, expected_candidate_set_fingerprint: str | None = None, + session_id: str | None = None, ) -> dict[str, Any] | None: """Run the real allocator over the live queue for *request*'s scope. An incomplete candidate inventory returns ``None`` rather than a ranking over a partial set (#758): selecting from a short list can pick the wrong work unit, so the request denies instead. + + *session_id* is supplied by the caller so the dry-run and the apply share one + control-plane identity; the lease an apply creates is then owned by an id the + request flow still holds. A dry run additionally goes through the allocator's + ``side_effect_free`` path, so a preview writes no session row and sweeps no + leases (#643). """ import control_plane_db @@ -917,7 +1162,7 @@ def default_allocator( db = control_plane_db.ControlPlaneDB() result = allocator_service.allocate_next_work( db, - session_id=f"webui-request-{uuid.uuid4().hex[:12]}", + session_id=session_id or _new_session_id(), role=request.desired_role, remote=request.remote, org=request.org, @@ -926,6 +1171,7 @@ def default_allocator( apply=bool(apply), allocation_mode="role_scoped", expected_candidate_set_fingerprint=expected_candidate_set_fingerprint, + side_effect_free=not apply, ) if isinstance(result, dict): result.setdefault("candidate_count", len(candidates))