diff --git a/examples/dsh-agents-as-a-service/gateway/README.md b/examples/dsh-agents-as-a-service/gateway/README.md index 295fc01..625e360 100644 --- a/examples/dsh-agents-as-a-service/gateway/README.md +++ b/examples/dsh-agents-as-a-service/gateway/README.md @@ -121,7 +121,7 @@ can also be injected into `create_app` for tests. | `COOKIE_SECURE` | `false` | Add `Secure` to the session cookie when served over TLS. | | `COOKIE_MAX_AGE` | `604800` | Session cookie lifetime in seconds. | | `MAX_BODY_MB` | `32` | Maximum buffered request body (and `413` threshold). | -| `HTTP_TIMEOUT` | `300` | Upstream connect/read timeout in seconds. | +| `HTTP_TIMEOUT` | `300` | Upstream connect/read timeout in seconds. Proxied streams (SSE, long responses) keep the connect/write budget but have **no read timeout**, so an idle event stream is not cut every five minutes. | | `SWEEP_INTERVAL` | `15` | Seconds between idle sweeps. | ### Required RBAC diff --git a/examples/dsh-agents-as-a-service/gateway/gateway.py b/examples/dsh-agents-as-a-service/gateway/gateway.py index 7429c86..ba8a1a2 100644 --- a/examples/dsh-agents-as-a-service/gateway/gateway.py +++ b/examples/dsh-agents-as-a-service/gateway/gateway.py @@ -1049,7 +1049,11 @@ against the JSON API. The browser session is an HttpOnly cookie.

for attempt in range(self.settings.upstream_attempts): try: upstream_request = self.http.build_request( - request.method, url, headers=headers, content=body + request.method, + url, + headers=headers, + content=body, + timeout=self._stream_timeout(), ) return await self.http.send(upstream_request, stream=True) except (httpx.ConnectError, httpx.ConnectTimeout) as exc: @@ -1062,6 +1066,21 @@ against the JSON API. The browser session is an HttpOnly cookie.

raise UpstreamError("upstream request failed") from exc raise UpstreamError("upstream unavailable") from last_error + def _stream_timeout(self) -> httpx.Timeout: + """Proxied streams (SSE, long responses) must not hit a read timeout. + + An idle event stream would otherwise raise ``httpx.ReadTimeout`` every + ``HTTP_TIMEOUT`` seconds and cut the browser's connection. The relay + is cancelled when the browser disconnects and ``_relay`` closes the + upstream response, so an unbounded read cannot outlive the client. + """ + return httpx.Timeout( + connect=self.settings.http_timeout, + read=None, + write=self.settings.http_timeout, + pool=self.settings.http_timeout, + ) + @staticmethod def _response_headers(upstream: httpx.Response, user: str): headers = [] @@ -1076,8 +1095,15 @@ against the JSON API. The browser session is an HttpOnly cookie.

async def _relay(self, user: str, upstream: httpx.Response): try: - async for chunk in upstream.aiter_raw(): - yield chunk + try: + async for chunk in upstream.aiter_raw(): + yield chunk + except httpx.HTTPError as exc: + # The upstream stream ended abnormally (reset, network change, + # deadline). The browser reconnects; one traceback per idle SSE + # stream is noise, so end the response cleanly instead. + print(f"stream for {user} ended early: {exc!r}", flush=True) + return finally: with contextlib.suppress(Exception): await upstream.aclose() diff --git a/examples/dsh-agents-as-a-service/gateway/test_gateway.py b/examples/dsh-agents-as-a-service/gateway/test_gateway.py index 898450c..977d5e8 100644 --- a/examples/dsh-agents-as-a-service/gateway/test_gateway.py +++ b/examples/dsh-agents-as-a-service/gateway/test_gateway.py @@ -1105,6 +1105,65 @@ async def test_proxy_cookie_mutation_requires_same_origin(env): assert good.status_code == 200 +def test_stream_timeout_leaves_read_unbounded(env): + timeout = env.gateway._stream_timeout() + assert timeout.read is None + assert timeout.connect == env.settings.http_timeout + + +@pytest.mark.asyncio +async def test_proxied_requests_use_the_stream_timeout(env, monkeypatch): + from starlette.requests import Request + + captured = {} + real_build = env.gateway.http.build_request + + def spy_build(method, url, **kwargs): + captured.update(kwargs) + return real_build(method, url, **kwargs) + + monkeypatch.setattr(env.gateway.http, "build_request", spy_build) + scope = { + "type": "http", + "method": "GET", + "scheme": "http", + "path": "/u/alice/events", + "raw_path": b"/u/alice/events", + "query_string": b"", + "root_path": "", + "headers": [(b"host", b"gateway.test")], + "client": ("127.0.0.1", 4321), + "server": ("gateway.test", 80), + "path_params": {"user": "alice", "path": "events"}, + } + response = await env.gateway._send_upstream( + Request(scope), "alice", FQDN, "dsh-auth-a=1", b"" + ) + await response.aclose() + + assert captured["timeout"].read is None + + +@pytest.mark.asyncio +async def test_relay_ends_cleanly_when_the_upstream_stream_breaks(env): + class Broken: + closed = False + + async def aiter_raw(self): + yield b"chunk" + raise httpx.ReadTimeout("upstream went quiet") + + async def aclose(self): + self.closed = True + + upstream = Broken() + chunks = [chunk async for chunk in env.gateway._relay("alice", upstream)] + + assert chunks == [b"chunk"] + assert upstream.closed is True + assert env.gateway.state.in_flight["alice"] == 0 + + # --- WebSocket proxy -------------------------------------------------------