From 2645014df55d74ea23f703cb4f7cd63d026943af Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Tue, 28 Jul 2026 18:01:11 -0600 Subject: [PATCH] fix(mux): add retry-after to listener refusals Both secure-listener 503 refusal paths were shipping without Retry-After, which AC2 requires as a 503 carrying Retry-After. Neither product code nor tests had it, so the gap was invisible to the suite. A live spot check over a real sandbox tunnel caught what fixtures missed: with the streaming lane full, the refused SSE consumer received 503 with Content-Type, Content-Length, and Connection: close, but no Retry-After. Use a single SECURE_LISTENER_REFUSAL_RETRY_AFTER_SECONDS constant set to 5 seconds so both paths stay aligned. Five seconds is short enough that transient saturation does not feel stale, and long enough that refused SSE consumers do not reconnect hard and recreate the storm this work exists to prevent. Tests assert against the constant instead of duplicating a literal. Also correct the observer-over-PL design doc's stale secure-listener identity line reference. --- docs/design/observer-over-pl.md | 2 +- solstone/convey/secure_listener/wsgi.py | 25 +++++++++++----- tests/link/test_mux.py | 39 ++++++++++++++++++++++++- 3 files changed, 57 insertions(+), 9 deletions(-) diff --git a/docs/design/observer-over-pl.md b/docs/design/observer-over-pl.md index 584bab6c0..3519d46bd 100644 --- a/docs/design/observer-over-pl.md +++ b/docs/design/observer-over-pl.md @@ -142,7 +142,7 @@ prefix for history, stats, and SSE registration. The helper reads `g.identity` directly. PL identity is already stamped by Convey: `install_identity_stamper()` sets DL identity defaults at `solstone/convey/__init__.py:95-108`, while the secure listener stamps -`request.environ["pl.identity"]` at `solstone/convey/secure_listener/wsgi.py:213-239`. +`request.environ["pl.identity"]` at `solstone/convey/secure_listener/wsgi.py:472`. The identity shape is `mode`, `fingerprint`, `device_label`, `paired_at`, and `session_id` (`solstone/convey/secure_listener/identity.py:12-18`). diff --git a/solstone/convey/secure_listener/wsgi.py b/solstone/convey/secure_listener/wsgi.py index 28793e4e4..ea021e827 100644 --- a/solstone/convey/secure_listener/wsgi.py +++ b/solstone/convey/secure_listener/wsgi.py @@ -34,6 +34,8 @@ _DEFAULT_PORTS: Final[dict[str, int]] = {"http": 80, "https": 443} WSGI_SEND_BRIDGE_POLL_SECONDS: Final[float] = 0.5 WSGI_INPUT_READ_TIMEOUT_SECONDS: Final[float] = 120.0 STREAMING_PERMIT_WAIT_TIMEOUT_SECONDS: Final[float] = 1.0 +# Five seconds dampens reconnect storms without making transient saturation stale. +SECURE_LISTENER_REFUSAL_RETRY_AFTER_SECONDS: Final[int] = 5 CAPACITY_SNAPSHOT_PATH: Final[str] = "/__solstone/secure-listener/capacity" _SAFE_STREAM_REFUSAL_METHODS: Final[frozenset[str]] = frozenset({"GET", "HEAD"}) # The cert-less pairing tunnel admits EXACTLY these endpoints (canonical + @@ -411,6 +413,12 @@ async def dispatch_stream( 503, "Service Unavailable", {"error": "secure listener capacity is full"}, + extra_headers=( + ( + "Retry-After", + str(SECURE_LISTENER_REFUSAL_RETRY_AFTER_SECONDS), + ), + ), ) stream_writer.begin_drain(RESET_CTX_BODY_DISCARD_CANCELLATION) return DispatchResult(endpoint=endpoint, status=503) @@ -513,16 +521,18 @@ async def write_json_response( payload: dict[str, Any], *, include_body: bool = True, + extra_headers: Iterable[tuple[str, str]] = (), ) -> None: body = json.dumps(payload, separators=(",", ":")).encode("utf-8") response_body = body if include_body else b"" - head = ( - f"HTTP/1.1 {status_code} {reason}\r\n" - "Content-Type: application/json\r\n" - f"Content-Length: {len(body)}\r\n" - "Connection: close\r\n" - "\r\n" - ).encode("ascii") + header_lines = [ + f"HTTP/1.1 {status_code} {reason}\r\n", + "Content-Type: application/json\r\n", + f"Content-Length: {len(body)}\r\n", + ] + header_lines.extend(f"{name}: {value}\r\n" for name, value in extra_headers) + header_lines.extend(("Connection: close\r\n", "\r\n")) + head = "".join(header_lines).encode("ascii") await writer.write(head + response_body) await writer.close() @@ -736,6 +746,7 @@ def _send_refusal_response(send: Callable[[bytes], None], method: str) -> None: "HTTP/1.1 503 Service Unavailable\r\n" "Content-Type: application/json\r\n" f"Content-Length: {len(body)}\r\n" + f"Retry-After: {SECURE_LISTENER_REFUSAL_RETRY_AFTER_SECONDS}\r\n" "Connection: close\r\n" "\r\n" ).encode("ascii") diff --git a/tests/link/test_mux.py b/tests/link/test_mux.py index 351165855..01ab453ef 100644 --- a/tests/link/test_mux.py +++ b/tests/link/test_mux.py @@ -12,6 +12,7 @@ import time from collections.abc import Iterator from dataclasses import dataclass from pathlib import Path +from types import SimpleNamespace from typing import Any import pytest @@ -1999,6 +2000,39 @@ async def test_content_length_json_response_does_not_take_streaming_permit( await _shutdown_admission(admission) +@pytest.mark.asyncio +async def test_ordinary_capacity_refusal_carries_retry_after_when_enabled( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + app, _journal = make_convey_app(tmp_path, monkeypatch, link={"posture": "spl"}) + fingerprint = "sha256:" + ("c" * 64) + _authorize_fingerprint(monkeypatch, fingerprint) + admission = _admission(capacity=1, streaming_capacity=0, refuse_when_full=True) + try: + with admission._lock: + admission._active_total = admission.config.capacity + for _ in range(admission.config.queue_limit): + admission._waiters.append(SimpleNamespace(queued_at=0.0)) + + status, headers, body, writer = await _dispatch_raw_request( + app, + pl_identity(fingerprint), + admission, + "GET", + "/app/network/api/status", + ) + + assert status == 503 + assert writer.closed is True + assert headers["retry-after"] == str( + wsgi_module.SECURE_LISTENER_REFUSAL_RETRY_AFTER_SECONDS + ) + assert b"secure listener capacity is full" in body + finally: + await _shutdown_admission(admission) + + @pytest.mark.asyncio async def test_get_streaming_response_refuses_before_body_when_lane_full( tmp_path: Path, @@ -2019,7 +2053,7 @@ async def test_get_streaming_response_refuses_before_body_when_lane_full( held = admission.try_acquire_streaming() assert held is not None try: - status, _headers, body, writer = await _dispatch_raw_request( + status, headers, body, writer = await _dispatch_raw_request( app, pl_identity(fingerprint), admission, @@ -2030,6 +2064,9 @@ async def test_get_streaming_response_refuses_before_body_when_lane_full( snapshot = admission.snapshot() assert status == 503 assert writer.closed is True + assert headers["retry-after"] == str( + wsgi_module.SECURE_LISTENER_REFUSAL_RETRY_AFTER_SECONDS + ) assert b"secure listener streaming capacity is full" in body assert b"should-not-send" not in body assert snapshot["rejected"]["streaming"] == 1 -- 2.51.2