From 05616607e48802e8e008927673c17b6ddaee27f7 Mon Sep 17 00:00:00 2001 From: Renn F Date: Tue, 30 Jun 2026 18:09:48 +0200 Subject: [PATCH] [chore] logical-gaps: release-proposal already_published closes proposal + heartbeat-lock-loss cancels execute (2 gaps) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit approve() closed the proposal only on status=='published'. A retry that finds the tag already shipped returns 'already_published' (is_already_published), so if a prior publish's route commit failed / HTTP 504'd, the proposal stayed non-terminal forever — every retry returned already_published and never closed it; only a manual cancel unstuck it. Close on both published and already_published: the release shipped either way. _heartbeat_loop returned silently when the lock was no longer owned (a >TTL Redis outage let the mutex expire mid-execute), leaving executor.execute running UNGUARDED — a concurrent approve (once Redis returns) could then acquire the lock and _prepare_release_clone rm -rf the in-flight shared release clone while the first execute was still mid-run_gate, re-opening the very rm -rf-clone race the mutex+heartbeat exist to prevent. Run execute as a task; on lock-loss the heartbeat sets a flag and cancels it, and approve() turns the CancelledError into a structured 'lock_lost' result (an external cancellation of approve itself still propagates — distinguished by the flag). TDD: 2 red→green (already_published → COMPLETED not wedged; heartbeat lock-loss → lock_lost + execute cancelled, proposal not completed). 8 concurrency tests green; ruff/mypy clean. --- roboco/services/release_proposal.py | 67 ++++++++++++-- .../test_release_proposal_concurrency.py | 91 +++++++++++++++++++ 2 files changed, 150 insertions(+), 8 deletions(-) diff --git a/roboco/services/release_proposal.py b/roboco/services/release_proposal.py index ea54fda3..8777dcfd 100644 --- a/roboco/services/release_proposal.py +++ b/roboco/services/release_proposal.py @@ -140,13 +140,46 @@ class ReleaseProposalService(BaseService): ) heartbeat_task: asyncio.Task[None] | None = None + execute_task: asyncio.Task[ReleaseResult] | None = None + # Set by the heartbeat when IT cancels execute on lock-loss, so the + # CancelledError handler below can distinguish a lock-loss abort (→ + # structured ``lock_lost`` result) from an external cancellation of the + # approve coroutine itself (→ must propagate). + lock_lost = asyncio.Event() try: executor = await get_release_executor(self.session) + execute_task = asyncio.create_task(executor.execute(report)) heartbeat_task = asyncio.create_task( - self._heartbeat_loop(lock_key, lock_token) + self._heartbeat_loop(lock_key, lock_token, execute_task, lock_lost) ) - result = await executor.execute(report) - if result.status == "published": + try: + result = await execute_task + except asyncio.CancelledError: + if not lock_lost.is_set(): + # External cancellation of approve itself — propagate, do not + # mask it as a lock-loss. + raise + logger.critical( + "release execute aborted: lock lost mid-execute (fail-closed)" + ) + return ReleaseResult( + status="lock_lost", + version=report.proposed_version, + files_changed=[], + commit_sha=None, + release_url=None, + detail=( + "The release lock was lost mid-execute (an extended Redis" + " outage let the mutex TTL expire); the execute was" + " aborted fail-closed so a concurrent approve could not" + " rm -rf the in-flight release clone. Retry the approve." + ), + ) + # Close the proposal when the release actually shipped — including a + # retry that finds the tag already published (a prior publish whose + # route commit/HTTP 504'd left the proposal non-terminal). The old + # `== "published"`-only check wedged already_published open forever. + if result.status in ("published", "already_published"): task.status = TaskStatus.COMPLETED await self.session.flush() return result @@ -154,6 +187,9 @@ class ReleaseProposalService(BaseService): if heartbeat_task is not None: heartbeat_task.cancel() await asyncio.gather(heartbeat_task, return_exceptions=True) + if execute_task is not None and not execute_task.done(): + execute_task.cancel() + await asyncio.gather(execute_task, return_exceptions=True) await self._release_release_lock(lock_key, lock_token) async def _acquire_release_lock(self, lock_key: str) -> str | None: @@ -205,22 +241,37 @@ class ReleaseProposalService(BaseService): finally: await conn.aclose() - async def _heartbeat_loop(self, lock_key: str, token: str) -> None: + async def _heartbeat_loop( + self, + lock_key: str, + token: str, + execute_task: asyncio.Task[ReleaseResult], + lock_lost: asyncio.Event, + ) -> None: """Refresh the lock TTL while the execute owns it. Refreshes before the first sleep so a fast execute still extends the TTL. A refresh error logs and continues (never crashes the execute); if - the lock is no longer ours (returned 0) we stop — the TTL backstop and - the fencing token still hold the line. + the lock is no longer ours (returned 0 — only reachable after a >TTL + Redis outage lets the mutex expire mid-execute) we CANCEL the in-flight + execute fail-closed rather than ``return`` silently and leave it running + unguarded — otherwise a concurrent approve (once Redis returns) can + acquire the lock and ``_prepare_release_clone`` ``rm -rf``'s the shared + release clone while the first execute is still mid-``run_gate``. The + fencing token still prevents the first finally from deleting the + usurper's lock; this prevents the usurper's rm -rf from corrupting the + first execute. """ while True: try: if not await self._heartbeat_release_lock(lock_key, token): logger.critical( "release lock no longer owned during execute — " - "TTL backstop active; a concurrent approve was refused " - "by the fencing token" + "aborting execute fail-closed so a concurrent approve" + " cannot rm -rf the in-flight release clone" ) + lock_lost.set() + execute_task.cancel() return except Exception as exc: logger.warning("release lock heartbeat failed (redis): %s", exc) diff --git a/tests/unit/services/test_release_proposal_concurrency.py b/tests/unit/services/test_release_proposal_concurrency.py index 21e2e113..57632ba8 100644 --- a/tests/unit/services/test_release_proposal_concurrency.py +++ b/tests/unit/services/test_release_proposal_concurrency.py @@ -393,3 +393,94 @@ async def test_heartbeat_refreshes_lock_and_is_cancelled_in_finally( assert all(k == name for k, _ttl in fake_redis.expire_calls) # And the lock was released at the end. assert await fake_redis.get(name) is None + + +@pytest.mark.asyncio +async def test_already_published_closes_proposal_not_wedges_open() -> None: + """#149: when a prior publish landed (tag exists) but the route commit + failed / 504'd, a retry sees the tag and execute returns ``already_published``. + approve() must STILL mark the proposal COMPLETED — the release shipped — + else the old ``if status == 'published'`` check wedged it open forever (every + retry returns already_published and never closes it; only a manual cancel + unsticks it).""" + task = _task() + fake_redis = _FakeRedis() + already = ReleaseResult( + status="already_published", + version="0.13.0", + files_changed=[], + commit_sha=None, + release_url=None, + detail="v0.13.0 is already published; nothing to do.", + ) + w = _wire(task, _REPORT, already, fake_redis) + svc = ReleaseProposalService(_session()) + + with ( + w["patches"][0], + w["patches"][1], + w["patches"][2], + w["patches"][3], + w["patches"][4], + ): + result = await svc.approve(task.id) + + assert result is not None + assert result.status == "already_published" + # The proposal is CLOSED — the release shipped — not wedged open. + assert task.status == TaskStatus.COMPLETED.value + + +@pytest.mark.asyncio +async def test_heartbeat_lock_loss_cancels_execute_fail_closed( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """#218: when the heartbeat finds the lock no longer owned (a >TTL Redis + outage let the mutex expire and a usurper re-acquired), it must CANCEL the + in-flight execute fail-closed — not ``return`` silently and leave execute + running unguarded, which lets the usurper's approve ``rm -rf`` the in-flight + release clone (re-opening the race the mutex+heartbeat exist to prevent). + approve() surfaces a structured ``lock_lost`` result.""" + monkeypatch.setattr(rp, "_RELEASE_LOCK_HEARTBEAT_SECONDS", 0.001) + + task = _task() + fake_redis = _FakeRedis() + published = ReleaseResult( + status="published", + version="0.13.0", + files_changed=[], + commit_sha=None, + release_url=None, + detail="ok", + ) + + execute_started = asyncio.Event() + + async def _blocking_execute(_report: Any) -> ReleaseResult: + # Block until the heartbeat cancels us — proves execute was running and + # got cancelled, not never-started. + execute_started.set() + await asyncio.sleep(60) + return published + + w = _wire(task, _REPORT, published, fake_redis) + w["executor"].execute = AsyncMock(side_effect=_blocking_execute) + svc = ReleaseProposalService(_session()) + # The heartbeat's compare-and-expire reports the lock lost (token mismatch — + # a usurper re-acquired after TTL expiry). + svc._heartbeat_release_lock = AsyncMock(return_value=False) + + with ( + w["patches"][0], + w["patches"][1], + w["patches"][2], + w["patches"][3], + w["patches"][4], + ): + result = await svc.approve(task.id) + + assert result is not None + assert result.status == "lock_lost" + assert execute_started.is_set() # execute did start, then was cancelled + # Fail-closed: the proposal is NOT marked COMPLETED. + assert task.status != TaskStatus.COMPLETED.value