fix(gateway): answer client-aborted requests with 499, not a traceback (#32)

* fix(gateway): answer client-aborted requests with 499, not a traceback

A browser reload, back navigation or proxy timeout can abort a request while
its body is still arriving. Starlette then raises ClientDisconnect from
request.stream(); nothing caught it, so ServerErrorMiddleware logged a full
ERROR traceback per aborted request and buried real faults. The request never
reaches DSH (the body is buffered before the upstream send) and the peer can
no longer be answered, so this is not an error.

- Register a ClientDisconnect exception handler that answers 499
  (CLIENT_CLOSED_REQUEST). One handler covers the proxy and the body-reading
  control routes (signup, login). A request already streaming keeps its
  upstream status because the response has started.
- Make _release await-free, and reach it even when the upstream close is
  cancelled. CancelledError is a BaseException, so suppress(Exception) around
  upstream.aclose() skipped the release and stranded the in-flight slot,
  leaving that user's sandbox looking busy forever (the idle sweeper then
  never suspends it).
- Add raw-ASGI regression tests: mid-body abort on the proxy and on signup,
  mid-stream abort, a cancelled release, and a cancelled upstream close.
- Document the 499 semantics in the gateway README failure table and the
  architecture failure modes.

Verified: make check, make test (114 passed), make lint. Against the pinned
uvicorn 0.54 the abort reproduces the original traceback before the fix and
logs no traceback after it.

* chore(deps): bump gateway starlette to 1.10.0 and ruff to 0.17.2

Absorbs Dependabot PR #27 against current main: that branch predates the
dsh-v0.2.1-rc.5 changelog/release commits, so merging it as-is would have
reverted them.

starlette 1.10.0 reworks URL-encoded form parsing (fields are built in the
parser callbacks, and the last field is kept when the stream ends without an
empty chunk) and clears tracebacks of handled exceptions. The gateway reads
JSON and form bodies, so the existing suite exercises signup/login forms and
the 499 client-abort path against the new release. ruff 0.17.2 only adds
preview and fix-safety changes; the configured rule set reports nothing new.

Verified: make check, make test (114 passed), make lint, make gateway (image
builds on python:3.12-slim and imports starlette 1.10.0), plus a direct ASGI
abort probe under 1.10.0 (499, in-flight slot released, nothing forwarded).

---------

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

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

@@ -8,6 +8,7 @@ use Starlette's ``TestClient`` (which runs the lifespan) against a fake
 """
 
 import asyncio
+import contextlib
 import hashlib
 import json
 import threading
@@ -263,6 +264,43 @@ def stale_handshake(env):
     env.gateway.state.handshake[("alice", POD_UID, "gateway.test")] = "dsh-auth-a=stale"
 
 
+def raw_scope(method, path, headers=()):
+    """An ASGI HTTP scope for driving the app without an httpx client.
+
+    ``spec_version`` matches what uvicorn 0.54 advertises (2.3), which is what
+    puts ``StreamingResponse`` on its disconnect-listening path.
+    """
+    return {
+        "type": "http",
+        "asgi": {"version": "3.0", "spec_version": "2.3"},
+        "http_version": "1.1",
+        "method": method,
+        "scheme": "http",
+        "path": path,
+        "raw_path": path.encode(),
+        "query_string": b"",
+        "root_path": "",
+        "server": ("gateway.test", 80),
+        "client": ("10.0.0.9", 4321),
+        "headers": [(b"host", b"gateway.test"), *headers],
+    }
+
+
+async def asgi_call(env, scope, receive):
+    """Drive the app with a scripted ``receive`` and return the sent messages.
+
+    An exception escaping the app fails the test, which is exactly how a
+    ``ClientDisconnect``-induced traceback used to surface.
+    """
+    sent = []
+
+    async def send(message):
+        sent.append(message)
+
+    await env.app(scope, receive, send)
+    return sent
+
+
 # --- pure helpers ----------------------------------------------------------
 
 
@@ -1049,6 +1087,162 @@ async def test_proxy_body_limit_returns_413(env):
     assert env.gateway.state.in_flight["alice"] == 0
 
 
+@pytest.mark.asyncio
+async def test_proxy_client_disconnect_mid_body_returns_499(env):
+    """An aborted upload is answered 499 quietly and never reaches DSH."""
+    add_user(env.k8s)
+    messages = [
+        {"type": "http.request", "body": b"partial", "more_body": True},
+        {"type": "http.disconnect"},
+    ]
+
+    async def receive():
+        return messages.pop(0)
+
+    sent = await asgi_call(
+        env,
+        raw_scope(
+            "POST",
+            "/u/alice/api/echo",
+            headers=[
+                (b"origin", b"http://gateway.test"),
+                (b"cookie", f"dsh_gw_session=alice:{TOKEN}".encode()),
+                (b"content-type", b"application/json"),
+            ],
+        ),
+        receive,
+    )
+    starts = [message for message in sent if message["type"] == "http.response.start"]
+    assert [start["status"] for start in starts] == [499]
+    assert env.upstream.proxy_requests == []
+    assert env.gateway.state.in_flight["alice"] == 0
+
+
+@pytest.mark.asyncio
+async def test_signup_client_disconnect_mid_body_returns_499(env):
+    """Body-reading control routes get the same treatment, not just the proxy."""
+    messages = [
+        {"type": "http.request", "body": b'{"user": "ali', "more_body": True},
+        {"type": "http.disconnect"},
+    ]
+
+    async def receive():
+        return messages.pop(0)
+
+    sent = await asgi_call(
+        env,
+        raw_scope("POST", "/users", headers=[(b"content-type", b"application/json")]),
+        receive,
+    )
+    starts = [message for message in sent if message["type"] == "http.response.start"]
+    assert [start["status"] for start in starts] == [499]
+    assert env.k8s.claims == {}
+
+
+@pytest.mark.asyncio
+async def test_proxy_client_disconnect_mid_stream_frees_slot(env):
+    """Closing an open SSE-style stream frees the slot without a traceback."""
+    add_user(env.k8s)
+    first_chunk = asyncio.Event()
+
+    class SlowStream(httpx.AsyncByteStream):
+        async def __aiter__(self):
+            yield b"chunk-one"
+            first_chunk.set()
+            await asyncio.Event().wait()  # the upstream stream stays open
+
+    env.upstream.proxy_handler = lambda request: httpx.Response(
+        200, headers={"content-type": "text/event-stream"}, stream=SlowStream()
+    )
+    messages = [{"type": "http.request", "body": b"", "more_body": False}]
+
+    async def receive():
+        if messages:
+            return messages.pop(0)
+        await first_chunk.wait()
+        return {"type": "http.disconnect"}
+
+    sent = await asgi_call(
+        env,
+        raw_scope(
+            "GET",
+            "/u/alice/api/stream",
+            headers=[(b"cookie", f"dsh_gw_session=alice:{TOKEN}".encode())],
+        ),
+        receive,
+    )
+    starts = [message for message in sent if message["type"] == "http.response.start"]
+    assert [start["status"] for start in starts] == [200]
+    assert env.gateway.state.in_flight["alice"] == 0
+
+
+@pytest.mark.asyncio
+async def test_release_frees_slot_even_when_cancelled(env):
+    """A cancelled request must still return its in-flight slot.
+
+    Starlette cancels the streaming task through an anyio cancel scope when the
+    client disconnects, and the cancellation can land inside ``_relay``'s
+    ``finally``.  Waiting for ``state.lock`` there would strand the slot, which
+    keeps the sandbox looking busy forever.
+    """
+    env.gateway.state.in_flight["alice"] = 1
+    await env.gateway.state.lock.acquire()
+    try:
+        task = asyncio.create_task(env.gateway._release("alice"))
+        await asyncio.sleep(0)
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+    finally:
+        env.gateway.state.lock.release()
+    assert env.gateway.state.in_flight["alice"] == 0
+
+
+@pytest.mark.asyncio
+async def test_proxy_client_disconnect_survives_cancelled_upstream_close(env):
+    """A cancellation that lands inside ``upstream.aclose()`` must not skip the release.
+
+    CancelledError is a BaseException, so ``contextlib.suppress(Exception)``
+    around the close would let it escape before ``_release`` ran.  A real
+    network close awaits, which is where the cancel scope delivers it.
+    """
+    add_user(env.k8s)
+    first_chunk = asyncio.Event()
+
+    class ClosingStream(httpx.AsyncByteStream):
+        async def __aiter__(self):
+            yield b"chunk-one"
+            first_chunk.set()
+            await asyncio.Event().wait()
+
+        async def aclose(self):
+            await asyncio.sleep(0)  # a checkpoint for the cancel scope
+
+    env.upstream.proxy_handler = lambda request: httpx.Response(
+        200, headers={"content-type": "text/event-stream"}, stream=ClosingStream()
+    )
+    messages = [{"type": "http.request", "body": b"", "more_body": False}]
+
+    async def receive():
+        if messages:
+            return messages.pop(0)
+        await first_chunk.wait()
+        return {"type": "http.disconnect"}
+
+    sent = await asgi_call(
+        env,
+        raw_scope(
+            "GET",
+            "/u/alice/api/stream",
+            headers=[(b"cookie", f"dsh_gw_session=alice:{TOKEN}".encode())],
+        ),
+        receive,
+    )
+    starts = [message for message in sent if message["type"] == "http.response.start"]
+    assert [start["status"] for start in starts] == [200]
+    assert env.gateway.state.in_flight["alice"] == 0
+
+
 @pytest.mark.asyncio
 async def test_proxy_wakes_suspended_sandbox(env, monkeypatch):
     sandbox_name, _ = add_user(env.k8s, ready=False)