src/available_issue_snapshot_store.py

@@ -174,6 +174,16 @@ class AvailableIssueSnapshotStore:
             connection.commit()
         return self.load()
 
+    def release_refresh(self, owner: str | None) -> None:
+        if owner is None:
+            return
+        with self._connect() as connection:
+            connection.execute(
+                "DELETE FROM available_issue_refresh_lease "
+                "WHERE singleton = 1 AND owner = ?",
+                (owner,),
+            )
+
     def fail_refresh(self, owner: str | None, *, retry_at: float) -> AvailableIssueSnapshotState:
         now = self.clock()
         with self._connect() as connection:

src/main.py

@@ -1387,6 +1387,9 @@ async def _refresh_available_issue_snapshot(lease_owner: str) -> list[dict]:
         _available_issue_snapshot_retry_at = None
         return shared.items or []
     except asyncio.CancelledError:
+        await asyncio.to_thread(
+            _available_issue_snapshot_store.release_refresh, lease_owner
+        )
         raise
     except Exception:
         try:

tests/test_available_issue_snapshot_store.py

@@ -88,3 +88,14 @@ def test_catalog_store_is_private(tmp_path):
 
     assert os.stat(path.parent).st_mode & 0o777 == 0o700
     assert os.stat(path).st_mode & 0o777 == 0o600
+
+
+def test_cancelled_refresh_releases_its_lease_for_immediate_takeover(tmp_path):
+    path = tmp_path / "available.sqlite3"
+    first = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    second = AvailableIssueSnapshotStore(path, clock=lambda: 100.0)
+    owner = first.try_acquire_refresh(lease_seconds=30)
+
+    first.release_refresh(owner)
+
+    assert second.try_acquire_refresh(lease_seconds=30) is not None