Do not read-time-out proxied upstream streams (#9)

Every proxied response is streamed through _relay using one shared
httpx.AsyncClient configured with timeout=HTTP_TIMEOUT (300s). An idle SSE
or long-poll stream therefore raised httpx.ReadTimeout every five minutes,
which escaped the StreamingResponse as a full traceback and truncated the
browser's stream.

Build proxied requests with a timeout that leaves read unbounded (connect,
write and pool stay bounded), and end the relay quietly on httpx.HTTPError
instead of propagating it. The relay is cancelled when the browser
disconnects and _relay closes the upstream response, so an unbounded read
cannot outlive the client.

Docs: note the split in the HTTP_TIMEOUT row.

Co-authored-by: SREDevOpsAgents <agents@sredevops.org>

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

examples/dsh-agents-as-a-service/gateway/gateway.py

@@ -1049,7 +1049,11 @@ against the JSON API. The browser session is an HttpOnly cookie.</p>
         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.</p>
                 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.</p>
 
     async def _relay(self, user: str, upstream: httpx.Response):
         try:
-            async for chunk in upstream.aiter_raw():
+            try:
-                yield chunk
+                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()

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