diff --git a/solstone/convey/secure_listener/accept.py b/solstone/convey/secure_listener/accept.py index 466d9e61a..c55f92c78 100644 --- a/solstone/convey/secure_listener/accept.py +++ b/solstone/convey/secure_listener/accept.py @@ -13,6 +13,7 @@ import socket import uuid from collections.abc import Callable from dataclasses import dataclass, field +from functools import partial from typing import Any, Final, Literal from OpenSSL import SSL @@ -23,7 +24,13 @@ from solstone.think.link.window import window_open from .admission import SecureListenerAdmission from .framing import RESET_INTERNAL_ERROR from .identity import ConveyIdentity -from .mux import RESET_CTX_NO_IDENTITY, Multiplexer, ResetDiagnostic, StreamWriter +from .mux import ( + RESET_CTX_NO_IDENTITY, + RESET_CTX_SEND_CREDIT_STARVATION, + Multiplexer, + ResetDiagnostic, + StreamWriter, +) from .tls import TlsError, drive_tls, new_server from .wsgi import CERTLESS_PAIR_ENDPOINTS, DispatchResult, dispatch_stream @@ -315,30 +322,11 @@ class SecureListener: writer.stream_id, ) - def on_reset(diag: ResetDiagnostic) -> None: - self._log.info( - "secure stream reset conn=%s stream_id=%d reason=%s context=%s", - connection_id, - diag.stream_id, - diag.reason_name, - diag.context, - ) - self._emit( - "stream_reset", - { - "stream_id": diag.stream_id, - "reason_code": diag.reason_code, - "reason_name": diag.reason_name, - "context": diag.context, - "tunnel_id": connection_id, - }, - ) - mux = Multiplexer( send_frame, handle_stream, is_listener=True, - on_reset=on_reset, + on_reset=partial(self._on_stream_reset, connection_id), ) async def tcp_reader_loop() -> None: @@ -433,6 +421,36 @@ class SecureListener: await asyncio.gather(reader_task, writer_task, return_exceptions=True) await mux.close() + def _on_stream_reset(self, connection_id: str, diag: ResetDiagnostic) -> None: + if diag.context == RESET_CTX_SEND_CREDIT_STARVATION: + self._log.warning( + "secure stream starvation reset conn=%s stream_id=%d reason=%s " + "context=%s stall_age_ms=%.3f", + connection_id, + diag.stream_id, + diag.reason_name, + diag.context, + diag.stall_age_ms, + ) + else: + self._log.info( + "secure stream reset conn=%s stream_id=%d reason=%s context=%s", + connection_id, + diag.stream_id, + diag.reason_name, + diag.context, + ) + payload: dict[str, Any] = { + "stream_id": diag.stream_id, + "reason_code": diag.reason_code, + "reason_name": diag.reason_name, + "context": diag.context, + "tunnel_id": connection_id, + } + if diag.stall_age_ms is not None: + payload["stall_age_ms"] = diag.stall_age_ms + self._emit("stream_reset", payload) + def _identity_for_peer(self, mode: PeerMode, fingerprint: str) -> ConveyIdentity: entry = self._authorized.get(fingerprint) return ConveyIdentity( diff --git a/solstone/convey/secure_listener/mux.py b/solstone/convey/secure_listener/mux.py index 8d711cd9b..b339f01cb 100644 --- a/solstone/convey/secure_listener/mux.py +++ b/solstone/convey/secure_listener/mux.py @@ -21,6 +21,7 @@ from __future__ import annotations import asyncio import logging +import time from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Final, Protocol @@ -84,6 +85,7 @@ RESET_CTX_SEND_CREDIT_STARVATION: Final[str] = "send_credit_starvation" STREAM_CREDIT_STALL_TIMEOUT_SECONDS: Final[float] = 120.0 log = logging.getLogger("convey.secure_listener.mux") +_monotonic = time.monotonic _REASON_NAMES: Final[dict[int, str]] = { RESET_PROTOCOL_ERROR: "protocol_error", @@ -101,6 +103,7 @@ class ResetDiagnostic: reason_code: int reason_name: str context: str + stall_age_ms: float | None = None ResetDiag = Callable[[ResetDiagnostic], None] @@ -120,6 +123,7 @@ class _StreamState: recv_credit: int = INITIAL_WINDOW unacked_recv: int = 0 credit_event: asyncio.Event = field(default_factory=asyncio.Event) + send_credit_stall_started_at: float | None = None task: asyncio.Task[None] | None = None draining: bool = False drained_bytes: int = 0 @@ -145,6 +149,8 @@ class StreamWriter: ) chunk_len = min(len(view), RECOMMENDED_CHUNK, self._state.send_credit) if chunk_len <= 0: + if self._state.send_credit_stall_started_at is None: + self._state.send_credit_stall_started_at = _monotonic() self._state.credit_event.clear() try: await asyncio.wait_for( @@ -153,18 +159,24 @@ class StreamWriter: ) except TimeoutError as exc: if not self._state.writer_closed: + origin = self._state.send_credit_stall_started_at + stall_age_ms = (_monotonic() - origin) * 1000.0 # Withheld credit is ambiguous, not a protocol violation: # we locally cancel the stream to reclaim service capacity. await self._mux._emit_reset( self._state.stream_id, RESET_CANCEL, RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=stall_age_ms, ) self._mux._close_stream(self._state) raise ConnectionError( f"stream {self._state.stream_id} writer is closed" ) from exc continue + # A zero-credit WINDOW only wakes wait_for; the zero-credit condition + # ends when credit is actually consumed, so age may exceed the deadline. + self._state.send_credit_stall_started_at = None chunk = bytes(view[:chunk_len]) view = view[chunk_len:] self._state.send_credit -= chunk_len @@ -479,7 +491,14 @@ class Multiplexer: return await self._send_frame(encoded, urgent=urgent) - def _fire_diag(self, stream_id: int, reason: int, context: str) -> None: + def _fire_diag( + self, + stream_id: int, + reason: int, + context: str, + *, + stall_age_ms: float | None = None, + ) -> None: if self._on_reset is None: return if self._closed and context != RESET_CTX_SEND_CREDIT_STARVATION: @@ -491,18 +510,36 @@ class Multiplexer: reason_code=reason, reason_name=_REASON_NAMES.get(reason, "unspecified"), context=context, + stall_age_ms=stall_age_ms, ) try: self._on_reset(diag) except Exception: pass - async def _emit_reset(self, stream_id: int, reason: int, context: str) -> None: + async def _emit_reset( + self, + stream_id: int, + reason: int, + context: str, + *, + stall_age_ms: float | None = None, + ) -> None: if self._closed: - self._fire_diag(stream_id, reason, context) + self._fire_diag( + stream_id, + reason, + context, + stall_age_ms=stall_age_ms, + ) return await self._emit(build_reset(stream_id, reason)) - self._fire_diag(stream_id, reason, context) + self._fire_diag( + stream_id, + reason, + context, + stall_age_ms=stall_age_ms, + ) async def _reject_stream(self, frame: Frame, context: str) -> None: state = self._streams.get(frame.stream_id) diff --git a/tests/link/test_mux.py b/tests/link/test_mux.py index 01ab453ef..b1c7e0237 100644 --- a/tests/link/test_mux.py +++ b/tests/link/test_mux.py @@ -139,6 +139,14 @@ class _ObservedEvent(asyncio.Event): return await super().wait() +class _MutableClock: + def __init__(self, now: float) -> None: + self.now = now + + def __call__(self) -> float: + return self.now + + @dataclass class _ParkedWriter: mux: Multiplexer @@ -1677,11 +1685,195 @@ async def test_credit_stall_deadline_is_per_stall_and_window_resets_clock( await mux.close() +@pytest.mark.asyncio +async def test_send_credit_starvation_diagnostic_reports_stall_age_ms( + monkeypatch: pytest.MonkeyPatch, +) -> None: + deadline_seconds = 0.05 + clock = _MutableClock(5.0) + monkeypatch.setattr( + mux_module, "STREAM_CREDIT_STALL_TIMEOUT_SECONDS", deadline_seconds + ) + monkeypatch.setattr(mux_module, "_monotonic", clock) + diags: list[ResetDiagnostic] = [] + opened = asyncio.Event() + captured: dict[str, Any] = {} + state: Any | None = None + + async def send(_: bytes, *, urgent: bool = False) -> None: + return + + async def handler(reader: asyncio.StreamReader, writer: Any) -> None: + captured["reader"] = reader + captured["writer"] = writer + opened.set() + await asyncio.Event().wait() + + mux = Multiplexer(send, handler, is_listener=True, on_reset=diags.append) + try: + await mux.feed(build_open(1).encode()) + await asyncio.wait_for(opened.wait(), timeout=1.0) + state = mux._streams[1] + waiter_entered = asyncio.Event() + state.credit_event = _ObservedEvent(waiter_entered) + state.send_credit = 0 + + clock.now = 0.0 + write_task = asyncio.create_task(captured["writer"].write(b"x")) + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + clock.now = mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS + + with pytest.raises(ConnectionError, match="stream 1 writer is closed"): + await asyncio.wait_for(write_task, timeout=15.0) + assert diags == [ + ResetDiagnostic( + stream_id=1, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 1000.0, + ) + ] + finally: + if state is not None and state.task is not None and not state.task.done(): + state.task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await state.task + if open_state := mux._streams.get(1): + mux._close_stream(open_state) + await mux.close() + + +@pytest.mark.asyncio +async def test_credit_grant_resets_later_stall_origin( + monkeypatch: pytest.MonkeyPatch, +) -> None: + deadline_seconds = 0.05 + clock = _MutableClock(5.0) + monkeypatch.setattr( + mux_module, "STREAM_CREDIT_STALL_TIMEOUT_SECONDS", deadline_seconds + ) + monkeypatch.setattr(mux_module, "_monotonic", clock) + diags: list[ResetDiagnostic] = [] + sent_payloads: list[bytes] = [] + + async def send(data: bytes, *, urgent: bool = False) -> None: + for frame in _decode_frames([data]): + if frame.flags & FLAG_DATA: + sent_payloads.append(frame.payload) + + async def handler(*_: object) -> None: + await asyncio.Event().wait() + + mux = Multiplexer(send, handler, is_listener=True, on_reset=diags.append) + write_task: asyncio.Task[None] | None = None + try: + await mux.feed(build_open(1).encode()) + state = mux._streams[1] + waiter_entered = asyncio.Event() + state.credit_event = _ObservedEvent(waiter_entered) + state.send_credit = 0 + writer = StreamWriter(mux, state) + + clock.now = 0.0 + write_task = asyncio.create_task(writer.write(b"xy")) + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + + waiter_entered.clear() + await mux.feed(build_window(1, 1).encode()) + clock.now = 1.0 + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + clock.now = 1.0 + mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS + + with pytest.raises(ConnectionError, match="stream 1 writer is closed"): + await asyncio.wait_for(write_task, timeout=15.0) + assert sent_payloads == [b"x"] + assert len(diags) == 1 + assert diags[0].stream_id == 1 + assert diags[0].reason_code == RESET_CANCEL + assert diags[0].reason_name == "cancel" + assert diags[0].context == RESET_CTX_SEND_CREDIT_STARVATION + assert diags[0].stall_age_ms == pytest.approx( + mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 1000.0 + ) + finally: + if write_task is not None and not write_task.done(): + write_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await write_task + if state := mux._streams.get(1): + mux._close_stream(state) + await mux.close() + + +@pytest.mark.asyncio +async def test_zero_credit_window_does_not_reset_stall_origin( + monkeypatch: pytest.MonkeyPatch, +) -> None: + deadline_seconds = 0.05 + clock = _MutableClock(5.0) + monkeypatch.setattr( + mux_module, "STREAM_CREDIT_STALL_TIMEOUT_SECONDS", deadline_seconds + ) + monkeypatch.setattr(mux_module, "_monotonic", clock) + diags: list[ResetDiagnostic] = [] + + async def send(_: bytes, *, urgent: bool = False) -> None: + return + + async def handler(*_: object) -> None: + await asyncio.Event().wait() + + mux = Multiplexer(send, handler, is_listener=True, on_reset=diags.append) + write_task: asyncio.Task[None] | None = None + try: + await mux.feed(build_open(1).encode()) + state = mux._streams[1] + waiter_entered = asyncio.Event() + state.credit_event = _ObservedEvent(waiter_entered) + state.send_credit = 0 + writer = StreamWriter(mux, state) + + clock.now = 0.0 + write_task = asyncio.create_task(writer.write(b"x")) + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + + waiter_entered.clear() + await mux.feed(build_window(1, 0).encode()) + clock.now = mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 2 + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + + with pytest.raises(ConnectionError, match="stream 1 writer is closed"): + await asyncio.wait_for(write_task, timeout=15.0) + assert diags == [ + ResetDiagnostic( + stream_id=1, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 2000.0, + ) + ] + assert diags[0].stall_age_ms > ( + mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 1000.0 + ) + finally: + if write_task is not None and not write_task.done(): + write_task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await write_task + if state := mux._streams.get(1): + mux._close_stream(state) + await mux.close() + + @pytest.mark.asyncio async def test_send_credit_starvation_diagnostic_survives_closed_mux_without_storm( monkeypatch: pytest.MonkeyPatch, ) -> None: monkeypatch.setattr(mux_module, "STREAM_CREDIT_STALL_TIMEOUT_SECONDS", 0.05) + clock = _MutableClock(5.0) + monkeypatch.setattr(mux_module, "_monotonic", clock) diags: list[ResetDiagnostic] = [] opened = asyncio.Event() captured: dict[str, Any] = {} @@ -1704,19 +1896,22 @@ async def test_send_credit_starvation_diagnostic_survives_closed_mux_without_sto waiter_entered = asyncio.Event() state.credit_event = _ObservedEvent(waiter_entered) state.send_credit = 0 + clock.now = 0.0 write_task = asyncio.create_task(captured["writer"].write(b"x")) await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) mux._closed = True + clock.now = mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS with pytest.raises(ConnectionError, match="stream 1 writer is closed"): - await asyncio.wait_for(write_task, timeout=1.0) + await asyncio.wait_for(write_task, timeout=15.0) assert diags == [ ResetDiagnostic( stream_id=1, reason_code=RESET_CANCEL, reason_name="cancel", context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 1000.0, ) ] finally: @@ -1729,6 +1924,179 @@ async def test_send_credit_starvation_diagnostic_survives_closed_mux_without_sto await mux.close() +@pytest.mark.asyncio +async def test_send_credit_starvation_timeout_then_mux_close_without_duplicate_diag( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(mux_module, "STREAM_CREDIT_STALL_TIMEOUT_SECONDS", 0.05) + clock = _MutableClock(5.0) + monkeypatch.setattr(mux_module, "_monotonic", clock) + sent: list[bytes] = [] + diags: list[ResetDiagnostic] = [] + opened = asyncio.Event() + captured: dict[str, Any] = {} + + async def send(data: bytes, *, urgent: bool = False) -> None: + sent.append(data) + + async def handler(reader: asyncio.StreamReader, writer: Any) -> None: + captured["reader"] = reader + captured["writer"] = writer + opened.set() + await asyncio.Event().wait() + + mux = Multiplexer(send, handler, is_listener=True, on_reset=diags.append) + try: + await mux.feed(build_open(1).encode()) + await asyncio.wait_for(opened.wait(), timeout=1.0) + state = mux._streams[1] + waiter_entered = asyncio.Event() + state.credit_event = _ObservedEvent(waiter_entered) + state.send_credit = 0 + + clock.now = 0.0 + write_task = asyncio.create_task(captured["writer"].write(b"x")) + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + clock.now = mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS + + with pytest.raises(ConnectionError, match="stream 1 writer is closed"): + await asyncio.wait_for(write_task, timeout=15.0) + await mux.close() + + frames = _decode_frames(sent) + _assert_single_reset(frames, 1, RESET_CANCEL) + assert diags == [ + ResetDiagnostic( + stream_id=1, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 1000.0, + ) + ] + finally: + await mux.close() + + +@pytest.mark.asyncio +async def test_send_credit_starvation_raising_on_reset_still_reclaims_stream( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(mux_module, "STREAM_CREDIT_STALL_TIMEOUT_SECONDS", 0.05) + clock = _MutableClock(5.0) + monkeypatch.setattr(mux_module, "_monotonic", clock) + sent: list[bytes] = [] + observed_diags: list[ResetDiagnostic] = [] + + async def send(data: bytes, *, urgent: bool = False) -> None: + sent.append(data) + + def on_reset(diag: ResetDiagnostic) -> None: + observed_diags.append(diag) + raise RuntimeError("emit failed") + + async def handler(*_: object) -> None: + await asyncio.Event().wait() + + mux = Multiplexer(send, handler, is_listener=True, on_reset=on_reset) + try: + await mux.feed(build_open(1).encode()) + state = mux._streams[1] + waiter_entered = asyncio.Event() + state.credit_event = _ObservedEvent(waiter_entered) + state.send_credit = 0 + writer = StreamWriter(mux, state) + + clock.now = 0.0 + write_task = asyncio.create_task(writer.write(b"x")) + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + clock.now = mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS + + with pytest.raises(ConnectionError, match="stream 1 writer is closed"): + await asyncio.wait_for(write_task, timeout=15.0) + + frames = _decode_frames(sent) + _assert_single_reset(frames, 1, RESET_CANCEL) + assert observed_diags == [ + ResetDiagnostic( + stream_id=1, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 1000.0, + ) + ] + assert state.writer_closed is True + assert 1 not in mux._streams + finally: + await mux.close() + + +@pytest.mark.asyncio +async def test_two_streams_can_starve_concurrently_with_independent_ages( + monkeypatch: pytest.MonkeyPatch, +) -> None: + monkeypatch.setattr(mux_module, "STREAM_CREDIT_STALL_TIMEOUT_SECONDS", 0.05) + clock = _MutableClock(5.0) + monkeypatch.setattr(mux_module, "_monotonic", clock) + diags: list[ResetDiagnostic] = [] + opened: dict[int, asyncio.Event] = {1: asyncio.Event(), 3: asyncio.Event()} + captured: dict[int, Any] = {} + + async def send(_: bytes, *, urgent: bool = False) -> None: + return + + async def handler(_reader: asyncio.StreamReader, writer: Any) -> None: + captured[writer.stream_id] = writer + opened[writer.stream_id].set() + await asyncio.Event().wait() + + mux = Multiplexer(send, handler, is_listener=True, on_reset=diags.append) + tasks: list[asyncio.Task[None]] = [] + try: + await mux.feed(build_open(1).encode()) + await mux.feed(build_open(3).encode()) + await asyncio.wait_for(opened[1].wait(), timeout=1.0) + await asyncio.wait_for(opened[3].wait(), timeout=1.0) + + for stream_id, origin in ((1, 0.0), (3, 1.0)): + state = mux._streams[stream_id] + waiter_entered = asyncio.Event() + state.credit_event = _ObservedEvent(waiter_entered) + state.send_credit = 0 + clock.now = origin + task = asyncio.create_task(captured[stream_id].write(b"x")) + tasks.append(task) + await asyncio.wait_for(waiter_entered.wait(), timeout=1.0) + + clock.now = 1.0 + mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS + for task in tasks: + with pytest.raises(ConnectionError, match=r"stream [13] writer is closed"): + await asyncio.wait_for(task, timeout=15.0) + + assert sorted(diag.stream_id for diag in diags) == [1, 3] + by_stream = {diag.stream_id: diag for diag in diags} + assert by_stream[1].reason_code == RESET_CANCEL + assert by_stream[1].reason_name == "cancel" + assert by_stream[1].context == RESET_CTX_SEND_CREDIT_STARVATION + assert by_stream[1].stall_age_ms == pytest.approx( + (1.0 + mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS) * 1000.0 + ) + assert by_stream[3].reason_code == RESET_CANCEL + assert by_stream[3].reason_name == "cancel" + assert by_stream[3].context == RESET_CTX_SEND_CREDIT_STARVATION + assert by_stream[3].stall_age_ms == pytest.approx( + mux_module.STREAM_CREDIT_STALL_TIMEOUT_SECONDS * 1000.0 + ) + finally: + for task in tasks: + if not task.done(): + task.cancel() + with contextlib.suppress(asyncio.CancelledError): + await task + await mux.close() + + @pytest.mark.asyncio async def test_receive_window_credit_returns_after_head_and_body_consumption( tmp_path: Path, @@ -2670,7 +3038,9 @@ async def test_reset_diagnostics_distinguish_contexts_and_are_privacy_clean( "reason_code", "reason_name", "context", + "stall_age_ms", } + assert diag.stall_age_ms is None values = [str(value) for value in diag.__dict__.values()] for secret in forbidden: assert all(secret not in value for value in values) diff --git a/tests/test_secure_listener_runtime.py b/tests/test_secure_listener_runtime.py index 7e1c2515e..d5c23a4f1 100644 --- a/tests/test_secure_listener_runtime.py +++ b/tests/test_secure_listener_runtime.py @@ -30,7 +30,16 @@ from solstone.convey.secure_listener.admission import ( SecureListenerAdmissionRejected, resolve_admission_config, ) -from solstone.convey.secure_listener.framing import build_ping +from solstone.convey.secure_listener.framing import ( + RESET_CANCEL, + RESET_PROTOCOL_ERROR, + build_ping, +) +from solstone.convey.secure_listener.mux import ( + RESET_CTX_SEND_CREDIT_STARVATION, + RESET_CTX_UNKNOWN_STREAM, + ResetDiagnostic, +) from solstone.think.link import client as link_client from solstone.think.link.ca import load_or_generate_ca from solstone.think.link.nonces import NONCE_TTL_SECONDS, NonceStore @@ -513,6 +522,189 @@ async def test_pump_connection_writer_failure_ends_connection( assert tcp_writer.writes +def test_on_stream_reset_payload_keys_and_log_shape( + caplog: pytest.LogCaptureFixture, +) -> None: + emitted: list[tuple[str, dict[str, object]]] = [] + listener = _listener() + listener._emit = lambda event, fields: emitted.append((event, dict(fields))) + starvation_diag = ResetDiagnostic( + stream_id=7, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=250.0, + ) + non_starvation_diag = ResetDiagnostic( + stream_id=9, + reason_code=RESET_PROTOCOL_ERROR, + reason_name="protocol_error", + context=RESET_CTX_UNKNOWN_STREAM, + ) + + with caplog.at_level(logging.INFO, logger="convey.secure_listener.accept"): + listener._on_stream_reset("conn-starved", starvation_diag) + listener._on_stream_reset("conn-normal", non_starvation_diag) + + records = [ + record + for record in caplog.records + if record.name == "convey.secure_listener.accept" + ] + assert records[0].levelno == logging.WARNING + assert ( + records[0].getMessage() + == "secure stream starvation reset conn=conn-starved stream_id=7 " + "reason=cancel context=send_credit_starvation stall_age_ms=250.000" + ) + assert records[1].levelno == logging.INFO + assert ( + records[1].getMessage() == "secure stream reset conn=conn-normal stream_id=9 " + "reason=protocol_error context=unknown_stream" + ) + + starvation_payload = emitted[0][1] + non_starvation_payload = emitted[1][1] + assert emitted[0][0] == "stream_reset" + assert emitted[1][0] == "stream_reset" + assert set(starvation_payload) == { + "stream_id", + "reason_code", + "reason_name", + "context", + "tunnel_id", + "stall_age_ms", + } + assert starvation_payload == { + "stream_id": 7, + "reason_code": RESET_CANCEL, + "reason_name": "cancel", + "context": RESET_CTX_SEND_CREDIT_STARVATION, + "tunnel_id": "conn-starved", + "stall_age_ms": 250.0, + } + assert set(non_starvation_payload) == { + "stream_id", + "reason_code", + "reason_name", + "context", + "tunnel_id", + } + assert non_starvation_payload == { + "stream_id": 9, + "reason_code": RESET_PROTOCOL_ERROR, + "reason_name": "protocol_error", + "context": RESET_CTX_UNKNOWN_STREAM, + "tunnel_id": "conn-normal", + } + forbidden = { + "sha256:", + "127.0.0.1", + "token", + "BEGIN CERTIFICATE", + } + for _event, payload in emitted: + values = [str(value) for value in payload.values()] + for secret in forbidden: + assert all(secret not in value for value in values) + + +def test_on_stream_reset_correlates_two_starved_streams_to_tunnels( + caplog: pytest.LogCaptureFixture, +) -> None: + emitted: list[tuple[str, dict[str, object]]] = [] + listener = _listener() + listener._emit = lambda event, fields: emitted.append((event, dict(fields))) + + with caplog.at_level(logging.WARNING, logger="convey.secure_listener.accept"): + listener._on_stream_reset( + "conn-a", + ResetDiagnostic( + stream_id=1, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=50.0, + ), + ) + listener._on_stream_reset( + "conn-b", + ResetDiagnostic( + stream_id=3, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=75.0, + ), + ) + + warnings = [ + record.getMessage() + for record in caplog.records + if record.name == "convey.secure_listener.accept" + and record.levelno == logging.WARNING + ] + assert warnings == [ + "secure stream starvation reset conn=conn-a stream_id=1 reason=cancel " + "context=send_credit_starvation stall_age_ms=50.000", + "secure stream starvation reset conn=conn-b stream_id=3 reason=cancel " + "context=send_credit_starvation stall_age_ms=75.000", + ] + assert emitted == [ + ( + "stream_reset", + { + "stream_id": 1, + "reason_code": RESET_CANCEL, + "reason_name": "cancel", + "context": RESET_CTX_SEND_CREDIT_STARVATION, + "tunnel_id": "conn-a", + "stall_age_ms": 50.0, + }, + ), + ( + "stream_reset", + { + "stream_id": 3, + "reason_code": RESET_CANCEL, + "reason_name": "cancel", + "context": RESET_CTX_SEND_CREDIT_STARVATION, + "tunnel_id": "conn-b", + "stall_age_ms": 75.0, + }, + ), + ] + + +def test_on_stream_reset_logs_warning_before_raising_emit( + caplog: pytest.LogCaptureFixture, +) -> None: + listener = _listener() + + def emit(_event: str, _fields: dict[str, object]) -> None: + raise RuntimeError("emit failed") + + listener._emit = emit + + with caplog.at_level(logging.WARNING, logger="convey.secure_listener.accept"): + with pytest.raises(RuntimeError, match="emit failed"): + listener._on_stream_reset( + "conn-raising", + ResetDiagnostic( + stream_id=1, + reason_code=RESET_CANCEL, + reason_name="cancel", + context=RESET_CTX_SEND_CREDIT_STARVATION, + stall_age_ms=50.0, + ), + ) + + assert ( + "secure stream starvation reset conn=conn-raising stream_id=1 " + "reason=cancel context=send_credit_starvation stall_age_ms=50.000" + ) in caplog.text + + @pytest.mark.asyncio async def test_certless_reap_tears_down_on_passive_expiry( tmp_path: Path,