From 4f6ba0302d2e4ab2b16bd4a66c55684712075838 Mon Sep 17 00:00:00 2001 From: timmy Date: Sat, 8 Aug 2026 00:34:10 +0000 Subject: [PATCH] feat: bound aggregate Gitea transport traffic (#246) --- README.md | 10 ++ src/gitea_proxy.py | 123 +++++++++++++++++++++- tests/test_gitea_transport.py | 192 ++++++++++++++++++++++++++++++++++ 3 files changed, 321 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 590e2d2..ab6dc9c 100644 --- a/README.md +++ b/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**, diff --git a/src/gitea_proxy.py b/src/gitea_proxy.py index d74e1c0..54206c8 100644 --- a/src/gitea_proxy.py +++ b/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 diff --git a/tests/test_gitea_transport.py b/tests/test_gitea_transport.py index 16a1888..85e8353 100644 --- a/tests/test_gitea_transport.py +++ b/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 = []