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