From 6d532dbf32f76c8eb859e242ff2a7cd8b0ad541b Mon Sep 17 00:00:00 2001 From: Jer Miller Date: Wed, 15 Jul 2026 11:47:38 -0600 Subject: [PATCH] feat(spl): bound browser-blob receives + lost-ACK retry dedupe proof Add bounded WebSocket reads with absolute deadlines and ciphertext progress windows, plus in-memory non-blocking admission gates for global and per-sender browser-blob receives. Wire relay dispatch and blob receive deadlines with saturation telemetry while keeping the wire format, on-disk state, MAX_CT_LEN, pairing path, and forwarding path unchanged; all guards are in-memory and reset to zero on process restart. Add the lost-ACK retry proof showing that a fresh-blob_id retry after duplicate prune resolves to the manifest-backed survivor as a duplicate, sends the correlated ACK, and does not recreate the pruned segment directory. Co-Authored-By: Claude Opus 4.8 (1M context) --- solstone/think/spl/admission.py | 57 ++ solstone/think/spl/blob_receiver.py | 180 ++++-- solstone/think/spl/health.py | 1 + solstone/think/spl/relay_client.py | 49 +- solstone/think/spl/ws_buffer.py | 143 +++++ tests/spl/test_admission.py | 65 ++ tests/spl/test_blob_receiver_ingest.py | 594 ++++++++++++++++++- tests/spl/test_relay_client_blob_dispatch.py | 2 +- tests/spl/test_ws_buffer.py | 94 ++- 9 files changed, 1126 insertions(+), 59 deletions(-) create mode 100644 solstone/think/spl/admission.py create mode 100644 tests/spl/test_admission.py diff --git a/solstone/think/spl/admission.py b/solstone/think/spl/admission.py new file mode 100644 index 000000000..5c27df739 --- /dev/null +++ b/solstone/think/spl/admission.py @@ -0,0 +1,57 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +"""Non-blocking in-memory admission gates for SPL blob receives.""" + +from __future__ import annotations + +_GLOBAL_ADMISSION_CEILING = 32 +_SENDER_ADMISSION_CEILING = 4 + + +class BlobAdmissionGate: + def __init__(self, *, global_ceiling: int, sender_ceiling: int) -> None: + self._global_ceiling = global_ceiling + self._sender_ceiling = sender_ceiling + self._global_count = 0 + self._sender_counts: dict[str, int] = {} + self._saturated_count = 0 + + @property + def global_count(self) -> int: + return self._global_count + + @property + def saturated_count(self) -> int: + return self._saturated_count + + def sender_count(self, fp: str) -> int: + return self._sender_counts.get(fp, 0) + + def active_senders(self) -> int: + return len(self._sender_counts) + + def try_acquire_global(self) -> bool: + if self._global_count < self._global_ceiling: + self._global_count += 1 + return True + self._saturated_count += 1 + return False + + def release_global(self) -> None: + self._global_count = max(0, self._global_count - 1) + + def try_acquire_sender(self, fp: str) -> bool: + count = self._sender_counts.get(fp, 0) + if count < self._sender_ceiling: + self._sender_counts[fp] = count + 1 + return True + self._saturated_count += 1 + return False + + def release_sender(self, fp: str) -> None: + count = self._sender_counts.get(fp, 0) + if count <= 1: + self._sender_counts.pop(fp, None) + return + self._sender_counts[fp] = count - 1 diff --git a/solstone/think/spl/blob_receiver.py b/solstone/think/spl/blob_receiver.py index 84fe5fe9e..6923f856f 100644 --- a/solstone/think/spl/blob_receiver.py +++ b/solstone/think/spl/blob_receiver.py @@ -25,8 +25,15 @@ from solstone.think.convey_client import ConveyClient from solstone.think.link.auth import AuthorizedClients, ClientEntry from solstone.think.link.paths import LinkState, authorized_clients_path from solstone.think.link.upload_key import load_upload_key +from solstone.think.spl.admission import BlobAdmissionGate +from solstone.think.spl.health import REASON_RELAY_ADMISSION_SATURATED from solstone.think.spl.hpke import open_auth -from solstone.think.spl.ws_buffer import BufferedWsReader +from solstone.think.spl.ws_buffer import ( + BufferedWsReader, + WsBufferClosed, + WsProgressTimeout, + WsReadTimeout, +) log = logging.getLogger(__name__) @@ -38,6 +45,11 @@ MAX_CT_LEN = 80 * 1024 * 1024 MAX_ENTRIES = 64 MAX_FILE_BYTES = 16 * 1024 * 1024 MAX_TOTAL_BYTES = 64 * 1024 * 1024 +_OFFER_DEADLINE_S = 5.0 +_ENC_DEADLINE_S = 5.0 +_CT_DEADLINE_S = 300.0 +_CT_WINDOW_S = 10.0 +_CT_MIN_BYTES_PER_WINDOW = 64 * 1024 _OFFER_MAGIC = b"SBO1" _READY_MAGIC = b"SBR1" @@ -55,6 +67,7 @@ BlobIngestPost = Callable[ [str, str, str, dict[str, Any], list[tuple[str, bytes, str]], str], Awaitable[dict[str, Any]] | dict[str, Any], ] +AdmissionEmit = Callable[[str, dict[str, Any]], None] @dataclass(frozen=True) @@ -87,8 +100,34 @@ async def receive_blob( ws: Any, *, ingest_post: BlobIngestPost | None = None, + gate: BlobAdmissionGate | None = None, + emit: AdmissionEmit | None = None, + offer_deadline_s: float | None = None, + enc_deadline_s: float | None = None, + ct_deadline_s: float | None = None, + ct_window_s: float | None = None, + ct_min_bytes_per_window: int | None = None, + time_source: Callable[[], float] | None = None, ) -> None: - header = await reader.read_exactly(OFFER_LEN) + offer_deadline_s = ( + offer_deadline_s if offer_deadline_s is not None else _OFFER_DEADLINE_S + ) + enc_deadline_s = enc_deadline_s if enc_deadline_s is not None else _ENC_DEADLINE_S + ct_deadline_s = ct_deadline_s if ct_deadline_s is not None else _CT_DEADLINE_S + ct_window_s = ct_window_s if ct_window_s is not None else _CT_WINDOW_S + ct_min_bytes_per_window = ( + ct_min_bytes_per_window + if ct_min_bytes_per_window is not None + else _CT_MIN_BYTES_PER_WINDOW + ) + + try: + header = await reader.read_exactly_bounded( + OFFER_LEN, deadline_s=offer_deadline_s, time_source=time_source + ) + except (WsReadTimeout, WsBufferClosed): + await _close_ws(ws) + return try: offer = _parse_offer(header) except ValueError: @@ -116,63 +155,98 @@ async def receive_blob( await _close_ws(ws) return - await _send_ready(ws, 0x00) - enc = await reader.read_exactly(ENC_LEN) - ct = await reader.read_exactly(offer.ct_len) - info = _blob_info(offer.sender_fp) - try: - opened = open_auth( - enc, - load_upload_key().private_key, - info, - bytes.fromhex(entry.pubkey_spki), - ct, - offer.header, - ) - except Exception as exc: - log.warning( - "blob open failed: blob_id=%s sender_fp=%s ct_len=%d type=%s", - offer.blob_id_hex, - offer.sender_fingerprint, - offer.ct_len, - type(exc).__name__, - ) - await _close_ws(ws) - return + sender_held = False + if gate is not None: + if not gate.try_acquire_sender(offer.sender_fingerprint): + if emit is not None: + emit( + "admission_saturated", + { + "reason": REASON_RELAY_ADMISSION_SATURATED, + "count": gate.saturated_count, + }, + ) + log.warning( + "blob receive rejected: reason=%s", REASON_RELAY_ADMISSION_SATURATED + ) + await _close_ws(ws) + return + sender_held = True try: - unpacked = _safe_unpack(opened.plaintext) - post = ingest_post or _default_ingest_post - response = post( - unpacked.day, - unpacked.segment, - unpacked.host, - unpacked.meta, - unpacked.files, - entry.observer_handle, - ) - if hasattr(response, "__await__"): - response = await response - status = _ack_status(response) - except Exception as exc: - log.warning( - "blob ingest failed: blob_id=%s sender_fp=%s type=%s", + await _send_ready(ws, 0x00) + try: + enc = await reader.read_exactly_bounded( + ENC_LEN, deadline_s=enc_deadline_s, time_source=time_source + ) + ct = await reader.read_exactly_progress( + offer.ct_len, + deadline_s=ct_deadline_s, + window_s=ct_window_s, + min_bytes_per_window=ct_min_bytes_per_window, + time_source=time_source, + ) + except (WsReadTimeout, WsProgressTimeout, WsBufferClosed): + await _close_ws(ws) + return + + info = _blob_info(offer.sender_fp) + try: + opened = open_auth( + enc, + load_upload_key().private_key, + info, + bytes.fromhex(entry.pubkey_spki), + ct, + offer.header, + ) + except Exception as exc: + log.warning( + "blob open failed: blob_id=%s sender_fp=%s ct_len=%d type=%s", + offer.blob_id_hex, + offer.sender_fingerprint, + offer.ct_len, + type(exc).__name__, + ) + await _close_ws(ws) + return + + try: + unpacked = _safe_unpack(opened.plaintext) + post = ingest_post or _default_ingest_post + response = post( + unpacked.day, + unpacked.segment, + unpacked.host, + unpacked.meta, + unpacked.files, + entry.observer_handle, + ) + if hasattr(response, "__await__"): + response = await response + status = _ack_status(response) + except Exception as exc: + log.warning( + "blob ingest failed: blob_id=%s sender_fp=%s type=%s", + offer.blob_id_hex, + offer.sender_fingerprint, + type(exc).__name__, + ) + await _close_ws(ws) + return + k_ack = opened.export(b"spl-blob-ack-v1", 32) + await ws.send(_ack(offer.blob_id, status, k_ack)) + log.info( + "blob accepted: blob_id=%s sender_fp=%s status=%#04x files=%d", offer.blob_id_hex, offer.sender_fingerprint, - type(exc).__name__, + status, + len(unpacked.files), ) await _close_ws(ws) - return - k_ack = opened.export(b"spl-blob-ack-v1", 32) - await ws.send(_ack(offer.blob_id, status, k_ack)) - log.info( - "blob accepted: blob_id=%s sender_fp=%s status=%#04x files=%d", - offer.blob_id_hex, - offer.sender_fingerprint, - status, - len(unpacked.files), - ) - await _close_ws(ws) + finally: + if sender_held and gate is not None: + gate.release_sender(offer.sender_fingerprint) def _parse_offer(header: bytes) -> Offer: diff --git a/solstone/think/spl/health.py b/solstone/think/spl/health.py index 2bfb4dd44..212b5bf5a 100644 --- a/solstone/think/spl/health.py +++ b/solstone/think/spl/health.py @@ -8,6 +8,7 @@ REASON_SERVICE_TOKEN_REJECTED = "service_token_rejected" REASON_RELAY_TUNNEL_REJECTED = "relay_tunnel_rejected" REASON_RELAY_TUNNEL_UNREACHABLE = "relay_tunnel_unreachable" REASON_LOCAL_PRIVATE_LISTENER_UNREACHABLE = "local_private_listener_unreachable" +REASON_RELAY_ADMISSION_SATURATED = "relay_admission_saturated" OFFLINE_TUNNEL_REASONS = frozenset( { diff --git a/solstone/think/spl/relay_client.py b/solstone/think/spl/relay_client.py index 5ed51246c..b20c0b6f8 100644 --- a/solstone/think/spl/relay_client.py +++ b/solstone/think/spl/relay_client.py @@ -20,10 +20,16 @@ import websockets from websockets.asyncio.client import ClientConnection from websockets.exceptions import ConnectionClosed, InvalidStatus, InvalidURI +from solstone.think.spl.admission import ( + _GLOBAL_ADMISSION_CEILING, + _SENDER_ADMISSION_CEILING, + BlobAdmissionGate, +) from solstone.think.spl.health import ( LINK_HEALTH_EVENT, REASON_HOME_MISSING_MOBILE, REASON_LOCAL_PRIVATE_LISTENER_UNREACHABLE, + REASON_RELAY_ADMISSION_SATURATED, REASON_RELAY_TUNNEL_REJECTED, REASON_RELAY_TUNNEL_UNREACHABLE, REASON_SERVICE_TOKEN_REJECTED, @@ -39,6 +45,7 @@ _HEALTH_REFRESH_SECONDS = 30.0 _LINK_DIRECT_HOST = "127.0.0.1" _LINK_DIRECT_PORT = 7657 _BUF = 65536 +_DISPATCH_READ_DEADLINE_S = 10.0 CallosumEmit = Callable[[str, dict[str, Any]], None] @@ -72,12 +79,20 @@ class RelayClient: relay_endpoint: str, service_token: str, callosum_emit: CallosumEmit | None = None, + dispatch_read_deadline_s: float = _DISPATCH_READ_DEADLINE_S, + global_admission_ceiling: int = _GLOBAL_ADMISSION_CEILING, + sender_admission_ceiling: int = _SENDER_ADMISSION_CEILING, ) -> None: self._instance_id = instance_id self._relay_endpoint = relay_endpoint.rstrip("/") self._relay_ws_endpoint = _to_ws(self._relay_endpoint) self._service_token = service_token self._emit = callosum_emit or (lambda _event, _fields: None) + self._dispatch_read_deadline_s = dispatch_read_deadline_s + self._admission_gate = BlobAdmissionGate( + global_ceiling=global_admission_ceiling, + sender_ceiling=sender_admission_ceiling, + ) self._running = False self._tunnels: dict[str, asyncio.Task[None]] = {} self._listen_generation = 0 @@ -164,6 +179,7 @@ class RelayClient: async def _handle_tunnel(self, tunnel_id: str) -> None: assert self._service_token is not None tcp_writer: asyncio.StreamWriter | None = None + global_held = False try: async with websockets.connect( self._url_for(f"/tunnel/{tunnel_id}", token=self._service_token), @@ -173,8 +189,17 @@ class RelayClient: self._record_tunnel_success() self._emit_health() reader = BufferedWsReader(ws) - prefix = await reader.peek(4) + if not self._admission_gate.try_acquire_global(): + self._emit_admission_saturated() + await ws.close() + return + global_held = True + prefix = await reader.peek_bounded( + 4, deadline_s=self._dispatch_read_deadline_s + ) if prefix[:1] == b"\x16": + self._admission_gate.release_global() + global_held = False try: tcp_reader, tcp_writer = await asyncio.open_connection( _LINK_DIRECT_HOST, @@ -203,7 +228,12 @@ class RelayClient: elif prefix == b"SBO1": from solstone.think.spl.blob_receiver import receive_blob - await receive_blob(reader, ws) + await receive_blob( + reader, + ws, + gate=self._admission_gate, + emit=self._emit, + ) else: log.info( "tunnel %s closed: unknown first bytes=%s", @@ -251,6 +281,8 @@ class RelayClient: except Exception as exc: # noqa: BLE001 log.warning("tunnel %s error: type=%s", tunnel_id, type(exc).__name__) finally: + if global_held: + self._admission_gate.release_global() if tcp_writer is not None: tcp_writer.close() with contextlib.suppress(OSError, RuntimeError): @@ -273,8 +305,21 @@ class RelayClient: "last_relay_tunnel_error": self._last_tunnel_error, "last_relay_tunnel_error_at": self._last_tunnel_error_at, "relay_tunnel_error_status": self._last_tunnel_error_status, + "relay_admission_saturated_count": ( + self._admission_gate.saturated_count + ), + }, + ) + + def _emit_admission_saturated(self) -> None: + self._emit( + "admission_saturated", + { + "reason": REASON_RELAY_ADMISSION_SATURATED, + "count": self._admission_gate.saturated_count, }, ) + log.warning("tunnel rejected: reason=%s", REASON_RELAY_ADMISSION_SATURATED) def _set_state(self, coarse_event: str, state: str) -> None: self._state = state diff --git a/solstone/think/spl/ws_buffer.py b/solstone/think/spl/ws_buffer.py index 15ff546aa..80f6610bb 100644 --- a/solstone/think/spl/ws_buffer.py +++ b/solstone/think/spl/ws_buffer.py @@ -5,6 +5,8 @@ from __future__ import annotations +import asyncio +from collections.abc import Callable from typing import Any from websockets.exceptions import ConnectionClosed @@ -14,6 +16,14 @@ class WsBufferClosed(RuntimeError): """Raised when a WebSocket closes before a requested byte count arrives.""" +class WsReadTimeout(TimeoutError): + """Raised when a bounded WebSocket read exceeds its absolute deadline.""" + + +class WsProgressTimeout(TimeoutError): + """Raised when a bounded WebSocket read does not make minimum progress.""" + + class BufferedWsReader: def __init__(self, ws: Any) -> None: self._ws = ws @@ -33,6 +43,54 @@ class BufferedWsReader: del self._buffer[:n] return out + async def peek_bounded( + self, + n: int, + *, + deadline_s: float, + time_source: Callable[[], float] | None = None, + ) -> bytes: + if n < 0: + raise ValueError("peek length must be non-negative") + await self._fill_bounded(n, deadline_s=deadline_s, time_source=time_source) + return bytes(self._buffer[:n]) + + async def read_exactly_bounded( + self, + n: int, + *, + deadline_s: float, + time_source: Callable[[], float] | None = None, + ) -> bytes: + if n < 0: + raise ValueError("read length must be non-negative") + await self._fill_bounded(n, deadline_s=deadline_s, time_source=time_source) + out = bytes(self._buffer[:n]) + del self._buffer[:n] + return out + + async def read_exactly_progress( + self, + n: int, + *, + deadline_s: float, + window_s: float, + min_bytes_per_window: int, + time_source: Callable[[], float] | None = None, + ) -> bytes: + if n < 0: + raise ValueError("read length must be non-negative") + await self._fill_bounded( + n, + deadline_s=deadline_s, + window_s=window_s, + min_bytes_per_window=min_bytes_per_window, + time_source=time_source, + ) + out = bytes(self._buffer[:n]) + del self._buffer[:n] + return out + def drain_buffer(self) -> bytes: out = bytes(self._buffer) self._buffer.clear() @@ -49,3 +107,88 @@ class BufferedWsReader: if not isinstance(frame, bytes): frame = frame.encode() self._buffer.extend(frame) + + async def _fill_bounded( + self, + n: int, + *, + deadline_s: float, + window_s: float | None = None, + min_bytes_per_window: int | None = None, + time_source: Callable[[], float] | None = None, + ) -> None: + if n < 0: + raise ValueError("fill length must be non-negative") + if (window_s is None) != (min_bytes_per_window is None): + raise ValueError("progress window requires both window and byte minimum") + if window_s is not None and window_s <= 0: + raise ValueError("progress window must be positive") + if min_bytes_per_window is not None and min_bytes_per_window < 0: + raise ValueError("minimum progress bytes must be non-negative") + + clock = ( + time_source if time_source is not None else asyncio.get_running_loop().time + ) + deadline = clock() + deadline_s + window_start = clock() + bytes_since_window_start = 0 + use_progress = window_s is not None and min_bytes_per_window is not None + + while len(self._buffer) < n: + now = clock() + if now >= deadline: + raise WsReadTimeout(f"websocket read exceeded {deadline_s:.3f}s") + + if ( + use_progress + and window_s is not None + and min_bytes_per_window is not None + ): + if now - window_start >= window_s: + if bytes_since_window_start < min_bytes_per_window: + raise WsProgressTimeout( + "websocket read progress below minimum " + f"({bytes_since_window_start} < {min_bytes_per_window} bytes)" + ) + window_start = now + bytes_since_window_start = 0 + timeout_at = min(deadline, window_start + window_s) + else: + timeout_at = deadline + + timeout = max(0.0, timeout_at - clock()) + try: + frame = await asyncio.wait_for(self._ws.recv(), timeout=timeout) + except asyncio.TimeoutError as exc: + now = clock() + if now >= deadline: + raise WsReadTimeout( + f"websocket read exceeded {deadline_s:.3f}s" + ) from exc + if ( + use_progress + and window_s is not None + and min_bytes_per_window is not None + and now - window_start >= window_s + ): + if bytes_since_window_start < min_bytes_per_window: + raise WsProgressTimeout( + "websocket read progress below minimum " + f"({bytes_since_window_start} < {min_bytes_per_window} bytes)" + ) from exc + window_start = now + bytes_since_window_start = 0 + continue + raise WsReadTimeout( + f"websocket read exceeded {deadline_s:.3f}s" + ) from exc + except ConnectionClosed as exc: + raise WsBufferClosed( + f"websocket closed before {n} bytes were available" + ) from exc + + if not isinstance(frame, bytes): + frame = frame.encode() + self._buffer.extend(frame) + if use_progress: + bytes_since_window_start += len(frame) diff --git a/tests/spl/test_admission.py b/tests/spl/test_admission.py new file mode 100644 index 000000000..c916e0497 --- /dev/null +++ b/tests/spl/test_admission.py @@ -0,0 +1,65 @@ +# SPDX-License-Identifier: AGPL-3.0-only +# Copyright (c) 2026 sol pbc + +from __future__ import annotations + +from solstone.think.spl.admission import BlobAdmissionGate + + +def test_global_ceiling_rejects_without_waiting_and_releases_capacity() -> None: + gate = BlobAdmissionGate(global_ceiling=2, sender_ceiling=4) + + assert gate.try_acquire_global() is True + assert gate.try_acquire_global() is True + assert gate.try_acquire_global() is False + assert gate.global_count == 2 + assert gate.saturated_count == 1 + + gate.release_global() + + assert gate.global_count == 1 + assert gate.try_acquire_global() is True + assert gate.global_count == 2 + assert gate.saturated_count == 1 + + +def test_per_sender_ceiling_is_keyed_by_fingerprint() -> None: + gate = BlobAdmissionGate(global_ceiling=10, sender_ceiling=1) + + assert gate.try_acquire_sender("sender-a") is True + assert gate.try_acquire_sender("sender-a") is False + assert gate.try_acquire_sender("sender-b") is True + assert gate.sender_count("sender-a") == 1 + assert gate.sender_count("sender-b") == 1 + assert gate.saturated_count == 1 + + gate.release_sender("sender-a") + + assert gate.try_acquire_sender("sender-a") is True + assert gate.sender_count("sender-a") == 1 + + +def test_sender_keys_are_pruned_at_zero() -> None: + gate = BlobAdmissionGate(global_ceiling=10, sender_ceiling=2) + + assert gate.try_acquire_sender("sender-a") is True + assert gate.try_acquire_sender("sender-a") is True + assert gate.active_senders() == 1 + + gate.release_sender("sender-a") + assert gate.sender_count("sender-a") == 1 + assert gate.active_senders() == 1 + + gate.release_sender("sender-a") + assert gate.sender_count("sender-a") == 0 + assert gate.active_senders() == 0 + + +def test_saturated_count_is_monotonic_across_global_and_sender_rejections() -> None: + gate = BlobAdmissionGate(global_ceiling=0, sender_ceiling=0) + + assert gate.try_acquire_global() is False + assert gate.try_acquire_sender("sender-a") is False + assert gate.try_acquire_global() is False + + assert gate.saturated_count == 3 diff --git a/tests/spl/test_blob_receiver_ingest.py b/tests/spl/test_blob_receiver_ingest.py index 4a8af0dfa..048bc36ac 100644 --- a/tests/spl/test_blob_receiver_ingest.py +++ b/tests/spl/test_blob_receiver_ingest.py @@ -3,7 +3,9 @@ from __future__ import annotations +import asyncio import gzip +import hashlib import hmac import io import json @@ -16,12 +18,17 @@ import pytest from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec +from solstone.apps.observer.prune import run_prune +from solstone.apps.observer.utils import append_history_record, observer_filename_prefix from solstone.observe.protocol import OBSERVER_HANDLE_HEADER from solstone.think.link.auth import AuthorizedClients from solstone.think.link.paths import LinkState, authorized_clients_path from solstone.think.link.upload_key import load_or_generate_upload_key from solstone.think.spl import blob_receiver, hpke +from solstone.think.spl.admission import BlobAdmissionGate +from solstone.think.spl.health import REASON_RELAY_ADMISSION_SATURATED from solstone.think.spl.ws_buffer import BufferedWsReader +from solstone.think.streams import write_segment_stream from tests._baseline_harness import mark_setup_complete @@ -45,6 +52,230 @@ class BlobWs: self.closed = True +class BlockingBlobWs(BlobWs): + def __init__(self, frames: list[bytes]) -> None: + super().__init__(frames) + self.release = asyncio.Event() + + async def recv(self) -> bytes: + if self.frames: + return self.frames.pop(0) + await self.release.wait() + from websockets.exceptions import ConnectionClosed + + raise ConnectionClosed(None, None) + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +class AdvancingBlobWs(BlobWs): + def __init__( + self, + frames: list[bytes], + clock: FakeClock, + advances: list[float], + ) -> None: + super().__init__(frames) + self.clock = clock + self.advances = list(advances) + + async def recv(self) -> bytes: + if self.frames: + if self.advances: + self.clock.advance(self.advances.pop(0)) + return self.frames.pop(0) + from websockets.exceptions import ConnectionClosed + + raise ConnectionClosed(None, None) + + +def _setup_browser_receiver( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, + *, + host: str = "browserhost", +) -> dict[str, Any]: + journal = tmp_path / "journal" + journal.mkdir() + mark_setup_complete(journal) + monkeypatch.setenv("SOLSTONE_JOURNAL", str(journal)) + + from solstone.convey import create_app + + app = create_app(journal=str(journal)) + client = app.test_client() + state = LinkState.load_or_create() + home_key = load_or_generate_upload_key() + ext_private = ec.generate_private_key(ec.SECP256R1()) + ext_spki = ext_private.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + sender_fp = hashes.Hash(hashes.SHA256()) + sender_fp.update(ext_spki) + sender_fp_bytes = sender_fp.finalize() + + register = client.post( + "/app/observer/register", + json={ + "platform": "browser", + "hostname": host, + "stream_type": "browser", + "version": "spl-browser-blob-v1", + }, + ) + assert register.status_code == 200 + observer_handle = register.get_json()["key"] + AuthorizedClients(authorized_clients_path()).add_browser( + fingerprint="sha256:" + sender_fp_bytes.hex(), + device_label=host, + instance_id=state.instance_id, + pubkey_spki=ext_spki.hex(), + observer_handle=observer_handle, + ) + + responses: list[dict[str, Any]] = [] + + async def ingest_post( + day: str, + segment: str, + got_host: str, + meta: dict[str, Any], + files: list[tuple[str, bytes, str]], + observer_handle_arg: str, + ) -> dict[str, Any]: + response = client.post( + "/app/observer/ingest", + headers={OBSERVER_HANDLE_HEADER: observer_handle_arg}, + data={ + "day": day, + "segment": segment, + "host": got_host, + "platform": "browser", + "meta": json.dumps(meta, separators=(",", ":")), + "files": [ + (io.BytesIO(content), filename) + for filename, content, _content_type in files + ], + }, + ) + assert response.status_code == 200 + body = response.get_json() + responses.append(body) + return body + + return { + "journal": journal, + "client": client, + "state": state, + "home_spki": home_key.public_spki_der, + "ext_private": ext_private, + "sender_fp_bytes": sender_fp_bytes, + "sender_fp": "sha256:" + sender_fp_bytes.hex(), + "observer_handle": observer_handle, + "ingest_post": ingest_post, + "responses": responses, + } + + +def _add_authorized_browser_sender( + *, + state: LinkState, + observer_handle: str, + host: str = "browserhost", +) -> tuple[ec.EllipticCurvePrivateKey, bytes]: + private = ec.generate_private_key(ec.SECP256R1()) + spki = private.public_key().public_bytes( + serialization.Encoding.DER, + serialization.PublicFormat.SubjectPublicKeyInfo, + ) + sender_fp = hashes.Hash(hashes.SHA256()) + sender_fp.update(spki) + sender_fp_bytes = sender_fp.finalize() + AuthorizedClients(authorized_clients_path()).add_browser( + fingerprint="sha256:" + sender_fp_bytes.hex(), + device_label=host, + instance_id=state.instance_id, + pubkey_spki=spki.hex(), + observer_handle=observer_handle, + ) + return private, sender_fp_bytes + + +def _write_browser_segment( + journal: Path, + *, + day: str, + stream: str, + segment: str, + seq: int, + prev: str | None, + filename: str, + content: bytes, +) -> Path: + seg_dir = journal / "chronicle" / day / stream / segment + seg_dir.mkdir(parents=True) + write_segment_stream(seg_dir, stream, day if prev else None, prev, seq) + (seg_dir / filename).write_bytes(content) + (seg_dir / "ingest.json").write_text( + json.dumps( + { + "schema_version": 1, + "requested_segment": segment, + "files": { + filename: { + "sha256": hashlib.sha256(content).hexdigest(), + "size": len(content), + } + }, + }, + indent=2, + sort_keys=True, + ) + + "\n", + encoding="utf-8", + ) + return seg_dir + + +def _append_browser_upload_history( + *, + prefix: str, + day: str, + stream: str, + segment: str, + filename: str, + content: bytes, +) -> None: + append_history_record( + prefix, + day, + { + "ts": 1, + "segment": segment, + "stream": stream, + "files": [ + { + "submitted": filename, + "written": filename, + "size": len(content), + "sha256": hashlib.sha256(content).hexdigest(), + "disposition": "written", + } + ], + }, + ) + + @pytest.mark.asyncio async def test_blob_receiver_ingests_and_acknowledges_duplicate_via_real_observer( tmp_path: Path, @@ -312,6 +543,340 @@ async def test_default_ingest_post_uses_loopback_upload( ] +@pytest.mark.asyncio +async def test_blob_offer_timeout_sends_no_ready_or_ack( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness = _setup_browser_receiver(tmp_path, monkeypatch) + gate = BlobAdmissionGate(global_ceiling=10, sender_ceiling=1) + ws = BlockingBlobWs([b"SBO1"]) + + await blob_receiver.receive_blob( + BufferedWsReader(ws), + ws, + gate=gate, + offer_deadline_s=0.01, + ) + + assert ws.sent == [] + assert ws.closed is True + assert gate.global_count == 0 + assert gate.sender_count(harness["sender_fp"]) == 0 + assert gate.active_senders() == 0 + + +@pytest.mark.asyncio +async def test_blob_enc_timeout_sends_ready_only_and_releases_sender( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness = _setup_browser_receiver(tmp_path, monkeypatch) + offer, _enc, _ct, _k_ack, _blob_id = _sealed_blob_parts( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + _browser_payload("enc-timeout", 10), + ) + gate = BlobAdmissionGate(global_ceiling=10, sender_ceiling=1) + ws = BlockingBlobWs([offer]) + + await blob_receiver.receive_blob( + BufferedWsReader(ws), + ws, + gate=gate, + ingest_post=harness["ingest_post"], + enc_deadline_s=0.01, + ) + + assert ws.sent == [b"SBR1\x01\x00"] + assert ws.closed is True + assert gate.sender_count(harness["sender_fp"]) == 0 + assert gate.active_senders() == 0 + + +@pytest.mark.asyncio +async def test_blob_ct_timeout_sends_ready_only_and_releases_sender( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness = _setup_browser_receiver(tmp_path, monkeypatch) + offer, enc, ct, _k_ack, _blob_id = _sealed_blob_parts( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + _browser_payload("ct-timeout", 11), + ) + gate = BlobAdmissionGate(global_ceiling=10, sender_ceiling=1) + ws = BlockingBlobWs([offer + enc + ct[:1]]) + + await blob_receiver.receive_blob( + BufferedWsReader(ws), + ws, + gate=gate, + ingest_post=harness["ingest_post"], + ct_deadline_s=0.01, + ) + + assert ws.sent == [b"SBR1\x01\x00"] + assert ws.closed is True + assert gate.sender_count(harness["sender_fp"]) == 0 + assert gate.active_senders() == 0 + + +@pytest.mark.asyncio +async def test_blob_ct_progress_timeout_with_drip_feed_releases_sender( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness = _setup_browser_receiver(tmp_path, monkeypatch) + offer, enc, ct, _k_ack, _blob_id = _sealed_blob_parts( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + _browser_payload("ct-progress-timeout", 12), + ) + clock = FakeClock() + gate = BlobAdmissionGate(global_ceiling=10, sender_ceiling=1) + ws = AdvancingBlobWs( + [offer + enc, ct[:1], ct[1:2]], + clock, + advances=[0.0, 1.1, 1.1], + ) + + await blob_receiver.receive_blob( + BufferedWsReader(ws), + ws, + gate=gate, + ingest_post=harness["ingest_post"], + ct_deadline_s=100.0, + ct_window_s=1.0, + ct_min_bytes_per_window=2, + time_source=clock, + ) + + assert ws.sent == [b"SBR1\x01\x00"] + assert ws.closed is True + assert gate.sender_count(harness["sender_fp"]) == 0 + assert gate.active_senders() == 0 + + +@pytest.mark.asyncio +async def test_blob_per_sender_cap_rejects_without_ready_and_prunes_keys( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness = _setup_browser_receiver(tmp_path, monkeypatch) + gate = BlobAdmissionGate(global_ceiling=10, sender_ceiling=1) + events: list[tuple[str, dict[str, Any]]] = [] + offer, enc, ct, _k_ack, _blob_id = _sealed_blob_parts( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + _browser_payload("held-a", 13), + segment="121000_300", + ) + held_ws = BlockingBlobWs([offer + enc + ct[:1]]) + held_task = asyncio.create_task( + blob_receiver.receive_blob( + BufferedWsReader(held_ws), + held_ws, + gate=gate, + ingest_post=harness["ingest_post"], + ct_deadline_s=30.0, + ) + ) + for _ in range(100): + if held_ws.sent and gate.sender_count(harness["sender_fp"]) == 1: + break + await asyncio.sleep(0) + assert held_ws.sent == [b"SBR1\x01\x00"] + assert gate.sender_count(harness["sender_fp"]) == 1 + + rejected_ws, _k_ack, _blob_id = _sealed_blob_ws( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + _browser_payload("rejected-a", 14), + segment="121500_300", + ) + await blob_receiver.receive_blob( + BufferedWsReader(rejected_ws), + rejected_ws, + gate=gate, + emit=lambda event, fields: events.append((event, dict(fields))), + ingest_post=harness["ingest_post"], + ) + assert rejected_ws.sent == [] + assert rejected_ws.closed is True + assert events == [ + ( + "admission_saturated", + {"reason": REASON_RELAY_ADMISSION_SATURATED, "count": 1}, + ) + ] + serialized_events = json.dumps(events) + assert harness["sender_fp"] not in serialized_events + assert "rejected-a" not in serialized_events + + sender_b_private, sender_b_fp = _add_authorized_browser_sender( + state=harness["state"], + observer_handle=harness["observer_handle"], + ) + sender_b_ws, sender_b_k_ack, sender_b_blob_id = _sealed_blob_ws( + harness["state"].instance_id, + harness["home_spki"], + sender_b_private, + sender_b_fp, + _browser_payload("sender-b", 15), + segment="122000_300", + ) + await blob_receiver.receive_blob( + BufferedWsReader(sender_b_ws), + sender_b_ws, + gate=gate, + ingest_post=harness["ingest_post"], + ) + assert sender_b_ws.sent[0] == b"SBR1\x01\x00" + assert sender_b_ws.sent[1][0:6] == b"SBA1\x01\x00" + assert sender_b_ws.sent[1][22:38] == _ack_tag( + sender_b_k_ack, 0x00, sender_b_blob_id + ) + + held_ws.release.set() + await held_task + assert gate.sender_count(harness["sender_fp"]) == 0 + assert gate.active_senders() == 0 + + accepted_ws, accepted_k_ack, accepted_blob_id = _sealed_blob_ws( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + _browser_payload("accepted-a", 16), + segment="122500_300", + ) + await blob_receiver.receive_blob( + BufferedWsReader(accepted_ws), + accepted_ws, + gate=gate, + ingest_post=harness["ingest_post"], + ) + assert accepted_ws.sent[0] == b"SBR1\x01\x00" + assert accepted_ws.sent[1][0:6] == b"SBA1\x01\x00" + assert accepted_ws.sent[1][22:38] == _ack_tag( + accepted_k_ack, 0x00, accepted_blob_id + ) + assert gate.sender_count(harness["sender_fp"]) == 0 + assert gate.active_senders() == 0 + + +@pytest.mark.asyncio +async def test_lost_ack_retry_resolves_duplicate_via_prune_survivor( + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + harness = _setup_browser_receiver(tmp_path, monkeypatch) + day = "20260704" + stream = "browserhost.browser" + filename = "browser_browserhost.jsonl" + content = _browser_payload("lost-ack", 17) + prefix = observer_filename_prefix({"key": harness["observer_handle"]}) + + canonical_dir = _write_browser_segment( + harness["journal"], + day=day, + stream=stream, + segment="120000_300", + seq=1, + prev=None, + filename=filename, + content=content, + ) + redundant_dir = _write_browser_segment( + harness["journal"], + day=day, + stream=stream, + segment="120000_301", + seq=2, + prev="120000_300", + filename=filename, + content=content, + ) + for segment in ("120000_300", "120000_301"): + _append_browser_upload_history( + prefix=prefix, + day=day, + stream=stream, + segment=segment, + filename=filename, + content=content, + ) + + result = run_prune(days=[day], stream=stream, execute=True) + + assert result.refusals == [] + assert [candidate.analysis.segment for candidate in result.deleted] == [ + "120000_301" + ] + assert not redundant_dir.exists() + assert (canonical_dir / filename).read_bytes() == content + assert (canonical_dir / "ingest.json").exists() + + retry_ws, retry_k_ack, blob_id_b3 = _sealed_blob_ws( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + content, + segment="120000_301", + ) + # This fresh B3 was not present in the direct-written/pruned setup; the retry + # still dedupes by content identity rather than blob id. + await blob_receiver.receive_blob( + BufferedWsReader(retry_ws), + retry_ws, + ingest_post=harness["ingest_post"], + ) + + assert retry_ws.sent[0] == b"SBR1\x01\x00" + assert retry_ws.sent[1][0:6] == b"SBA1\x01\x01" + assert retry_ws.sent[1][6:22] == blob_id_b3 + assert retry_ws.sent[1][22:38] == _ack_tag(retry_k_ack, 0x01, blob_id_b3) + assert harness["responses"][-1]["status"] == "duplicate" + assert harness["responses"][-1]["existing_segment"] == "120000_300" + assert not redundant_dir.exists() + + reused_ws, reused_k_ack, reused_blob_id = _sealed_blob_ws( + harness["state"].instance_id, + harness["home_spki"], + harness["ext_private"], + harness["sender_fp_bytes"], + content, + segment="120000_301", + blob_id=blob_id_b3, + ) + await blob_receiver.receive_blob( + BufferedWsReader(reused_ws), + reused_ws, + ingest_post=harness["ingest_post"], + ) + + assert reused_blob_id == blob_id_b3 + assert reused_ws.sent[0] == b"SBR1\x01\x00" + assert reused_ws.sent[1][0:6] == b"SBA1\x01\x01" + assert reused_ws.sent[1][6:22] == blob_id_b3 + assert reused_ws.sent[1][22:38] == _ack_tag(reused_k_ack, 0x01, blob_id_b3) + assert harness["responses"][-1]["status"] == "duplicate" + assert harness["responses"][-1]["existing_segment"] == "120000_300" + assert not redundant_dir.exists() + + def _sealed_blob_ws( instance_id: str, home_spki: bytes, @@ -321,8 +886,33 @@ def _sealed_blob_ws( *, host: str = "browserhost", segment: str = "120000_300", + blob_id: bytes | None = None, ) -> tuple[BlobWs, bytes, bytes]: - blob_id = uuid.uuid4().bytes + offer, enc, ct, k_ack, blob_id = _sealed_blob_parts( + instance_id, + home_spki, + ext_private, + sender_fp, + file_content, + host=host, + segment=segment, + blob_id=blob_id, + ) + return BlobWs([offer + enc + ct]), k_ack, blob_id + + +def _sealed_blob_parts( + instance_id: str, + home_spki: bytes, + ext_private: ec.EllipticCurvePrivateKey, + sender_fp: bytes, + file_content: bytes, + *, + host: str = "browserhost", + segment: str = "120000_300", + blob_id: bytes | None = None, +) -> tuple[bytes, bytes, bytes, bytes, bytes]: + blob_id = blob_id or uuid.uuid4().bytes plaintext = _tar_payload(file_content, host=host, segment=segment) info = b"spl-blob-v1" + uuid.UUID(instance_id).bytes + sender_fp suite = hpke._suite() @@ -337,7 +927,7 @@ def _sealed_blob_ws( ) ct = ctx.seal(plaintext, offer) assert len(ct) == len(plaintext) + 16 - return BlobWs([offer + enc + ct]), ctx.export(b"spl-blob-ack-v1", 32), blob_id + return offer, enc, ct, ctx.export(b"spl-blob-ack-v1", 32), blob_id def _offer(sender_fp: bytes, blob_id: bytes, ct_len: int) -> bytes: diff --git a/tests/spl/test_relay_client_blob_dispatch.py b/tests/spl/test_relay_client_blob_dispatch.py index dddcb1d80..1bad38c19 100644 --- a/tests/spl/test_relay_client_blob_dispatch.py +++ b/tests/spl/test_relay_client_blob_dispatch.py @@ -103,7 +103,7 @@ async def test_handle_tunnel_blob_branch_does_not_open_loopback( async def fail_open_connection(*_args: Any) -> tuple[FakeTcpReader, FakeTcpWriter]: raise AssertionError("blob branch must not open loopback") - async def fake_receive(reader, got_ws, *, ingest_post=None) -> None: + async def fake_receive(reader, got_ws, **_kwargs: Any) -> None: nonlocal called assert got_ws is ws assert await reader.read_exactly(4) == b"SBO1" diff --git a/tests/spl/test_ws_buffer.py b/tests/spl/test_ws_buffer.py index ee1450519..bd81ce760 100644 --- a/tests/spl/test_ws_buffer.py +++ b/tests/spl/test_ws_buffer.py @@ -3,10 +3,17 @@ from __future__ import annotations +import asyncio + import pytest from websockets.exceptions import ConnectionClosed -from solstone.think.spl.ws_buffer import BufferedWsReader, WsBufferClosed +from solstone.think.spl.ws_buffer import ( + BufferedWsReader, + WsBufferClosed, + WsProgressTimeout, + WsReadTimeout, +) class FakeWs: @@ -19,6 +26,41 @@ class FakeWs: return self.frames.pop(0) +class BlockingWs: + def __init__(self) -> None: + self.event = asyncio.Event() + + async def recv(self) -> bytes: + await self.event.wait() + return b"" + + +class FakeClock: + def __init__(self) -> None: + self.now = 0.0 + + def __call__(self) -> float: + return self.now + + def advance(self, seconds: float) -> None: + self.now += seconds + + +class AdvancingWs: + def __init__( + self, frames: list[bytes], clock: FakeClock, *, advance_s: float + ) -> None: + self.frames = list(frames) + self.clock = clock + self.advance_s = advance_s + + async def recv(self) -> bytes: + if not self.frames: + raise ConnectionClosed(None, None) + self.clock.advance(self.advance_s) + return self.frames.pop(0) + + @pytest.mark.asyncio async def test_peek_does_not_consume() -> None: reader = BufferedWsReader(FakeWs([b"abcdef"])) @@ -51,3 +93,53 @@ async def test_short_close_raises_clear_error() -> None: with pytest.raises(WsBufferClosed, match="before 4 bytes"): await reader.read_exactly(4) + + +@pytest.mark.asyncio +async def test_read_exactly_bounded_returns_single_frame_payload() -> None: + reader = BufferedWsReader(FakeWs([b"abcdef"])) + + assert await reader.read_exactly_bounded(6, deadline_s=10.0) == b"abcdef" + + +@pytest.mark.asyncio +async def test_read_exactly_bounded_stall_times_out() -> None: + reader = BufferedWsReader(BlockingWs()) + + with pytest.raises(WsReadTimeout): + await reader.read_exactly_bounded(1, deadline_s=0.01) + + +@pytest.mark.asyncio +async def test_read_exactly_progress_drip_feed_below_window_minimum() -> None: + clock = FakeClock() + reader = BufferedWsReader(AdvancingWs([b"a", b"b", b"c"], clock, advance_s=1.1)) + + with pytest.raises(WsProgressTimeout): + await reader.read_exactly_progress( + 3, + deadline_s=10.0, + window_s=1.0, + min_bytes_per_window=2, + time_source=clock, + ) + + +@pytest.mark.asyncio +async def test_read_exactly_progress_absolute_deadline_is_not_reset() -> None: + clock = FakeClock() + reader = BufferedWsReader(AdvancingWs([b"aa", b"bb", b"cc"], clock, advance_s=1.1)) + + with pytest.raises(WsReadTimeout): + await reader.read_exactly_progress( + 6, + deadline_s=2.0, + window_s=1.0, + min_bytes_per_window=2, + time_source=clock, + ) + + +def test_bounded_timeout_exceptions_are_timeout_errors() -> None: + assert issubclass(WsReadTimeout, TimeoutError) + assert issubclass(WsProgressTimeout, TimeoutError) -- 2.54.0