[chore] logical-gaps: release-proposal already_published closes proposal + heartbeat-lock-loss cancels execute (2 gaps)

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

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

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