Merge pull request 'Bound aggregate Gitea traffic and coalesce concurrent reads' (#247) from timmy/246-gitea-transport-bulkhead into main

README.md

@@ -68,6 +68,16 @@ API process is running and does not contact Gitea. GET `/readyz` is the
 readiness check: it validates the configured Gitea credentials and returns
 HTTP 503 with an error when Gitea is unavailable or authentication fails.
 
+The application-lifetime Gitea transport bounds aggregate upstream traffic and
+coalesces identical concurrent GETs without caching completed responses. By
+default, at most eight requests run at once, with one slot reserved for authored
+mutations so polling and detail-read bursts cannot starve comments, reviews, or
+other writes. Requests that cannot enter within 250 ms fail as retryable HTTP
+503 responses. Tune these limits with `GITEA_MAX_CONCURRENCY` (minimum `2`) and
+`GITEA_ADMISSION_TIMEOUT_SECONDS` (minimum `0.001`); keep the admission timeout
+below the route deadlines. Streaming diff reads share the same read capacity,
+while POST, PATCH, PUT, and DELETE requests are never coalesced.
+
 ## Offline mobile shell
 
 At phone widths, a persistent bottom task dock keeps **Work**, **Find**, **New**,

src/gitea_proxy.py

@@ -2,6 +2,7 @@ import asyncio
 import os
 import re
 import shlex
+from contextlib import asynccontextmanager
 from typing import Any
 from urllib.parse import urlsplit
 
@@ -12,7 +13,116 @@ GITEA_TOKEN = os.getenv("GITEA_TOKEN", "")
 REVIEW_DIFF_MAX_BYTES = 64 * 1024
 REVIEW_DIFF_MAX_LINES = 400
 AVAILABLE_ISSUE_PAGE_CONCURRENCY = 3
-_client: httpx.AsyncClient | None = None
+_client: "GiteaTransport | None" = None
+
+
+class GiteaOverloadedError(RuntimeError):
+    """Raised when bounded transport admission expires before capacity is available."""
+
+
+class GiteaTransport:
+    """Application-lifetime HTTP transport with single-flight concurrent reads."""
+
+    def __init__(
+        self,
+        *,
+        max_concurrency: int = 8,
+        admission_timeout: float = 0.25,
+        **kwargs,
+    ) -> None:
+        max_concurrency = max(2, max_concurrency)
+        admission_timeout = max(0.001, admission_timeout)
+        self._http = httpx.AsyncClient(base_url=GITEA_URL, timeout=10, **kwargs)
+        self.max_concurrency = max_concurrency
+        self.admission_timeout = admission_timeout
+        self._request_slots = asyncio.Semaphore(max(2, max_concurrency))
+        self._read_slots = asyncio.Semaphore(max(1, max_concurrency - 1))
+        self._reads: dict[tuple, asyncio.Task[httpx.Response]] = {}
+
+    @property
+    def is_closed(self) -> bool:
+        return self._http.is_closed
+
+    def _read_key(self, url: str, kwargs: dict) -> tuple:
+        params = tuple(httpx.QueryParams(kwargs.get("params", {})).multi_items())
+        headers = tuple(sorted(httpx.Headers(kwargs.get("headers", {})).multi_items()))
+        return url, params, headers
+
+    async def _acquire(self, semaphore: asyncio.Semaphore) -> None:
+        try:
+            await asyncio.wait_for(semaphore.acquire(), timeout=self.admission_timeout)
+        except TimeoutError as exc:
+            raise GiteaOverloadedError("Gitea transport is saturated") from exc
+
+    async def _perform_get(self, url: str, kwargs: dict) -> httpx.Response:
+        await self._acquire(self._read_slots)
+        try:
+            await self._acquire(self._request_slots)
+            try:
+                return await self._http.get(url, **kwargs)
+            finally:
+                self._request_slots.release()
+        finally:
+            self._read_slots.release()
+
+    async def get(self, url: str, **kwargs) -> httpx.Response:
+        key = self._read_key(url, kwargs)
+        task = self._reads.get(key)
+        if task is None:
+            task = asyncio.create_task(self._perform_get(url, kwargs))
+            self._reads[key] = task
+            task.add_done_callback(
+                lambda completed, request_key=key: (
+                    self._reads.pop(request_key, None)
+                    if self._reads.get(request_key) is completed else None
+                )
+            )
+        return await asyncio.shield(task)
+
+    async def _mutate(self, method: str, url: str, kwargs: dict) -> httpx.Response:
+        await self._acquire(self._request_slots)
+        try:
+            return await self._http.request(method, url, **kwargs)
+        finally:
+            self._request_slots.release()
+
+    async def post(self, url: str, **kwargs) -> httpx.Response:
+        return await self._mutate("POST", url, kwargs)
+
+    async def patch(self, url: str, **kwargs) -> httpx.Response:
+        return await self._mutate("PATCH", url, kwargs)
+
+    async def put(self, url: str, **kwargs) -> httpx.Response:
+        return await self._mutate("PUT", url, kwargs)
+
+    async def delete(self, url: str, **kwargs) -> httpx.Response:
+        return await self._mutate("DELETE", url, kwargs)
+
+    @asynccontextmanager
+    async def stream(self, method: str, url: str, **kwargs):
+        await self._acquire(self._read_slots)
+        try:
+            await self._acquire(self._request_slots)
+            try:
+                async with self._http.stream(method, url, **kwargs) as response:
+                    yield response
+            finally:
+                self._request_slots.release()
+        finally:
+            self._read_slots.release()
+
+    def __getattr__(self, name: str):
+        return getattr(self._http, name)
+
+    async def aclose(self) -> None:
+        tasks = list(self._reads.values())
+        for task in tasks:
+            if not task.done():
+                task.cancel()
+        if tasks:
+            await asyncio.gather(*tasks, return_exceptions=True)
+        self._reads.clear()
+        await self._http.aclose()
 
 
 class WorkItems(list[dict]):
@@ -58,14 +168,19 @@ def _auth() -> dict[str, str]:
     return headers
 
 
-def start_client(**kwargs) -> httpx.AsyncClient:
+def start_client(**kwargs) -> GiteaTransport:
     """Create the application-lifetime Gitea transport."""
     global _client
-    _client = httpx.AsyncClient(base_url=GITEA_URL, timeout=10, **kwargs)
+    kwargs.setdefault("max_concurrency", int(os.getenv("GITEA_MAX_CONCURRENCY", "8")))
+    kwargs.setdefault(
+        "admission_timeout",
+        float(os.getenv("GITEA_ADMISSION_TIMEOUT_SECONDS", "0.25")),
+    )
+    _client = GiteaTransport(**kwargs)
     return _client
 
 
-def _get_client() -> httpx.AsyncClient:
+def _get_client() -> GiteaTransport:
     if _client is None or _client.is_closed:
         return start_client()
     return _client

tests/test_gitea_transport.py

@@ -28,6 +28,198 @@ async def test_gitea_transport_is_reused_across_requests_and_closed():
     assert client.is_closed
 
 
+@pytest.mark.anyio
+async def test_gitea_transport_coalesces_identical_concurrent_gets():
+    calls = 0
+    release = asyncio.Event()
+
+    async def handler(request):
+        nonlocal calls
+        calls += 1
+        await release.wait()
+        return httpx.Response(200, json={"path": request.url.path})
+
+    gitea_proxy.start_client(
+        transport=httpx.MockTransport(handler),
+        max_concurrency=4,
+        admission_timeout=0.1,
+    )
+    try:
+        requests = [asyncio.create_task(gitea_proxy.fetch("user")) for _ in range(20)]
+        await asyncio.sleep(0)
+        release.set()
+        results = await asyncio.gather(*requests)
+    finally:
+        await gitea_proxy.stop_client()
+
+    assert calls == 1
+    assert results == [{"path": "/api/v1/user"}] * 20
+
+
+@pytest.mark.anyio
+async def test_cancelling_one_coalesced_waiter_keeps_the_shared_get_running():
+    started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def handler(request):
+        started.set()
+        await release.wait()
+        return httpx.Response(200, json={"path": request.url.path})
+
+    gitea_proxy.start_client(transport=httpx.MockTransport(handler))
+    owner = asyncio.create_task(gitea_proxy.fetch("user"))
+    try:
+        await started.wait()
+        waiter = asyncio.create_task(gitea_proxy.fetch("user"))
+        await asyncio.sleep(0)
+        waiter.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await waiter
+        release.set()
+        assert await owner == {"path": "/api/v1/user"}
+    finally:
+        release.set()
+        await asyncio.gather(owner, return_exceptions=True)
+        await gitea_proxy.stop_client()
+
+
+@pytest.mark.anyio
+async def test_gitea_transport_rejects_reads_beyond_bounded_capacity():
+    active = 0
+    peak = 0
+    saturated = asyncio.Event()
+    release = asyncio.Event()
+
+    async def handler(request):
+        nonlocal active, peak
+        active += 1
+        peak = max(peak, active)
+        if active == 2:
+            saturated.set()
+        try:
+            await release.wait()
+            return httpx.Response(200, json={"path": request.url.path})
+        finally:
+            active -= 1
+
+    gitea_proxy.start_client(
+        transport=httpx.MockTransport(handler),
+        max_concurrency=3,
+        admission_timeout=0.02,
+    )
+    first = asyncio.create_task(gitea_proxy.fetch("first"))
+    second = asyncio.create_task(gitea_proxy.fetch("second"))
+    try:
+        await asyncio.wait_for(saturated.wait(), timeout=0.2)
+        with pytest.raises(gitea_proxy.GiteaOverloadedError):
+            await gitea_proxy.fetch("third")
+    finally:
+        release.set()
+        await asyncio.gather(first, second)
+        await gitea_proxy.stop_client()
+
+    assert peak == 2
+
+
+@pytest.mark.anyio
+async def test_gitea_transport_reserves_one_slot_for_mutations_and_bounds_them():
+    active = 0
+    peak = 0
+    reads_started = asyncio.Event()
+    mutation_started = asyncio.Event()
+    release = asyncio.Event()
+
+    async def handler(request):
+        nonlocal active, peak
+        active += 1
+        peak = max(peak, active)
+        if active == 2:
+            reads_started.set()
+        if request.method == "POST":
+            mutation_started.set()
+        try:
+            await release.wait()
+            return httpx.Response(200, json={"method": request.method})
+        finally:
+            active -= 1
+
+    client = gitea_proxy.start_client(
+        transport=httpx.MockTransport(handler),
+        max_concurrency=3,
+        admission_timeout=0.02,
+    )
+    reads = [
+        asyncio.create_task(gitea_proxy.fetch("first")),
+        asyncio.create_task(gitea_proxy.fetch("second")),
+    ]
+    mutation = None
+    try:
+        await asyncio.wait_for(reads_started.wait(), timeout=0.2)
+        mutation = asyncio.create_task(client.post("/api/v1/mutate", json={"value": 1}))
+        await asyncio.wait_for(mutation_started.wait(), timeout=0.2)
+        with pytest.raises(gitea_proxy.GiteaOverloadedError):
+            await asyncio.wait_for(
+                client.post("/api/v1/mutate", json={"value": 1}), timeout=0.1
+            )
+    finally:
+        release.set()
+        await asyncio.gather(*reads, *(task for task in [mutation] if task is not None))
+        await gitea_proxy.stop_client()
+
+    assert peak == 3
+
+
+@pytest.mark.anyio
+async def test_gitea_transport_reads_bulkhead_configuration_from_environment(monkeypatch):
+    monkeypatch.setenv("GITEA_MAX_CONCURRENCY", "12")
+    monkeypatch.setenv("GITEA_ADMISSION_TIMEOUT_SECONDS", "0.75")
+
+    client = gitea_proxy.start_client(
+        transport=httpx.MockTransport(lambda request: httpx.Response(200, json={}))
+    )
+    try:
+        assert client.max_concurrency == 12
+        assert client.admission_timeout == 0.75
+    finally:
+        await gitea_proxy.stop_client()
+
+
+@pytest.mark.anyio
+async def test_streaming_reads_share_the_same_bounded_read_capacity():
+    active = 0
+    saturated = asyncio.Event()
+    release = asyncio.Event()
+
+    async def handler(request):
+        nonlocal active
+        active += 1
+        if active == 2:
+            saturated.set()
+        try:
+            await release.wait()
+            return httpx.Response(200, content=b"diff")
+        finally:
+            active -= 1
+
+    gitea_proxy.start_client(
+        transport=httpx.MockTransport(handler),
+        max_concurrency=3,
+        admission_timeout=0.02,
+    )
+    streams = [
+        asyncio.create_task(gitea_proxy.fetch_text("first", 100)),
+        asyncio.create_task(gitea_proxy.fetch_text("second", 100)),
+    ]
+    try:
+        await asyncio.wait_for(saturated.wait(), timeout=0.2)
+        with pytest.raises(gitea_proxy.GiteaOverloadedError):
+            await asyncio.wait_for(gitea_proxy.fetch("third"), timeout=0.1)
+    finally:
+        release.set()
+        await asyncio.gather(*streams)
+        await gitea_proxy.stop_client()
+
+
 @pytest.mark.anyio
 async def test_application_lifespan_opens_and_closes_gitea_transport(monkeypatch):
     calls = []