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))