diff --git a/pyproject.toml b/pyproject.toml index d43aab0..2f81e14 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -29,6 +29,9 @@ dependencies = [ [tool.uv.sources] missioncore-plugin-sdk = { path = "packages/plugin-sdk", editable = true } +[project.optional-dependencies] +perception-stream = ["grpcio>=1.76,<2"] + [project.scripts] k1link = "k1link.device_plugins.xgrids_k1.cli:app" missioncore-datasets = "k1link.datasets.cli:app" diff --git a/src/k1link/perception/streaming_grpc.py b/src/k1link/perception/streaming_grpc.py new file mode 100644 index 0000000..471eede --- /dev/null +++ b/src/k1link/perception/streaming_grpc.py @@ -0,0 +1,409 @@ +"""TLS gRPC candidate carrying the existing raw-first binary wire unchanged. + +One stream per resident profile. The trusted local controller issues one-shot, +short-lived capabilities AFTER choosing the input epoch. Network code cannot +activate models, renew leases, reset temporal stores or authorize a reconnect. +No insecure listener/channel, route upload, background source queue or retry. + +The gRPC method uses identity byte serialization (not generated protobufs): +each request is <=64 KiB of the existing MCI2 byte stream, each response one +bounded MCR3 envelope. Source/result semantics remain in the existing adapters. +""" + +from __future__ import annotations + +import asyncio +import hmac +import secrets +import socket +import threading +import time +from collections.abc import Callable, Iterable +from contextlib import suppress +from dataclasses import dataclass, field +from hashlib import sha256 +from typing import Any + +import grpc # type: ignore[import-untyped] # Upstream wheel does not ship type stubs. + +from k1link.compute.live_perception import LiveIngressEvent + +from . import streaming_wire as wire +from .realtime_contract import StreamStart +from .streaming_ingress import StreamingIngress +from .streaming_lifecycle import StreamingLifecycle +from .streaming_network import ( + BRIDGE_RESERVATION, + INPUT_CHUNK, + MAX_REPLY, + LatestReplies, + decode_reply, + encode_reply, +) + +SERVICE = "missioncore.perception.v1.BinaryStream" +METHOD = f"/{SERVICE}/Exchange" +OPTIONS = ( + ("grpc.enable_retries", 0), + ("grpc.http2.bdp_probe", 0), + # A fixed 1 MiB HTTP/2 receive window, not the default 64 KiB which + # serializes large observations into many WAN/SSH round trips. Keep BDP + # auto-growth disabled; native buffering remains part of the RSS gate. + ("grpc.http2.lookahead_bytes", wire.MAX_FRAGMENT), + ("grpc.http2.write_buffer_size", INPUT_CHUNK), + ("grpc.max_metadata_size", 4096), + ("grpc.max_concurrent_streams", 2), +) + + +@dataclass(frozen=True) +class StreamAccess: + epoch: StreamStart + session_id: str + session_generation: int + token: str = field(repr=False) + + def metadata(self) -> tuple[tuple[str, str], ...]: + return ( + ("authorization", "Bearer " + self.token), + ("x-stream-binding", wire.binding(self.epoch)), + ) + + +class GrpcStreamEndpoint: + def __init__( + self, + runtime: StreamingLifecycle, + consume: Callable[[LiveIngressEvent], None], + notice: Callable[[dict[str, Any]], None], + *, + idle_timeout: float = 2.0, + io_timeout: float = 0.25, + ) -> None: + if runtime.continuity is None: + raise ValueError("external stream requires a resumable resident runtime") + if not 0 < idle_timeout <= 30 or not 0 < io_timeout <= 2: + raise ValueError("invalid bounded network timeouts") + self.runtime, self.consume, self.notice = runtime, consume, notice + self.idle_timeout, self.io_timeout = idle_timeout, io_timeout + self.lock = threading.Lock() + self.grant: tuple[StreamAccess, bytes, float] | None = None + self.active: tuple[StreamStart, LatestReplies] | None = None + self.last: dict[str, Any] | None = None + self._quarantine_thread: threading.Thread | None = None + + def issue(self, epoch: StreamStart, session_id: str, session_generation: int) -> StreamAccess: + """Local control plane only. Delivery of this secret is outside data-plane.""" + wire.identifier(session_id) + wire.decimal(session_generation) + if session_generation < 1: + raise ValueError("invalid acquisition generation") + self.runtime.check_input(epoch, synchronizing=True) + with self.lock: + if self.active is not None: + raise wire.StreamWireError("previous network stream has not drained") + access = StreamAccess(epoch, session_id, session_generation, secrets.token_hex(32)) + # Do not retain the plaintext secret on the endpoint. + self.grant = ( + StreamAccess(epoch, session_id, session_generation, ""), + sha256(access.token.encode()).digest(), + time.monotonic() + 30, + ) + return access + + def _claim(self, metadata: Iterable[tuple[str, str | bytes]]) -> StreamAccess: + values: dict[str, str | bytes] = {} + for key, value in metadata: + if key in ("authorization", "x-stream-binding"): + if key in values: + raise wire.StreamWireError("stream authorization rejected") + values[key] = value + token = values.get("authorization") + if not isinstance(token, str) or len(token) != 71 or not token.startswith("Bearer "): + raise wire.StreamWireError("stream authorization rejected") + with self.lock: + grant = self.grant + if ( + grant is None + or self.active is not None + or time.monotonic() >= grant[2] + or values.get("x-stream-binding") != wire.binding(grant[0].epoch) + or not hmac.compare_digest(sha256(token[7:].encode()).digest(), grant[1]) + ): + raise wire.StreamWireError("stream authorization rejected") + self.runtime.check_input(grant[0].epoch, synchronizing=True) + self.grant = None + self.active = (grant[0].epoch, LatestReplies()) + return grant[0] + + def publish(self, epoch: StreamStart, sequence: int, payload: bytes) -> bool: + self.runtime.check_input(epoch) + with self.lock: + if self.active is None or self.active[0] != epoch: + return False + return self.active[1].put(sequence, payload) + + def finish_results(self, epoch: StreamStart) -> None: + """Graph controller signals output EOF AFTER all admitted work is accounted.""" + with self.lock: + if self.active is not None and self.active[0] == epoch: + self.active[1].finish() + + async def serve(self, address: str, *, certificate: bytes, private_key: bytes) -> Any: + if not certificate or not private_key: + raise ValueError("TLS identity is required") + server = grpc.aio.server( + options=OPTIONS + + ( + ("grpc.max_receive_message_length", INPUT_CHUNK), + ("grpc.max_send_message_length", MAX_REPLY), + ), + maximum_concurrent_rpcs=2, + compression=grpc.Compression.NoCompression, + ) + server.add_generic_rpc_handlers( + ( + grpc.method_handlers_generic_handler( + SERVICE, {"Exchange": grpc.stream_stream_rpc_method_handler(self.exchange)} + ), + ) + ) + port = server.add_secure_port( + address, grpc.ssl_server_credentials(((private_key, certificate),)) + ) + if port == 0: + raise RuntimeError("TLS stream listener failed to bind") + await server.start() + return server, port + + async def exchange(self, unused_iterator: Any, context: Any) -> None: + if b"ssl" not in context.auth_context().get("transport_security_type", ()): + await context.abort(grpc.StatusCode.UNAUTHENTICATED, "TLS required") + try: + access = self._claim(context.invocation_metadata()) + except (ValueError, RuntimeError): + await context.abort(grpc.StatusCode.UNAUTHENTICATED, "stream authorization rejected") + return + assert self.active is not None + outbox = self.active[1] + left = right = receiver = reservation = None + tasks: list[asyncio.Task[None]] = [] + try: + reservation = self.runtime.mailbox.reserve_ingress(BRIDGE_RESERVATION) + left, right = socket.socketpair() + for sock in (left, right): + sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 32 * 1024) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 32 * 1024) + sock.setblocking(False) + receiver = StreamingIngress( + right, + self.runtime, + access.session_id, + access.session_generation, + self.consume, + self.notice, + input_epoch=access.epoch, + fragment_timeout=self.io_timeout, + idle_timeout=self.idle_timeout, + ) + receiver.start() + await context.send_initial_metadata((("x-stream-binding", wire.binding(access.epoch)),)) + + async def read_input() -> None: + while True: + async with asyncio.timeout(self.idle_timeout): + raw = await context.read() + if raw is grpc.aio.EOF: + # Explicit wire End can already have closed the peer. + with suppress(OSError): + left.shutdown(socket.SHUT_WR) + return + if not isinstance(raw, bytes) or not 0 < len(raw) <= INPUT_CHUNK: + raise wire.StreamWireError("invalid network chunk") + async with asyncio.timeout(self.io_timeout): + await asyncio.get_running_loop().sock_sendall(left, raw) + + async def write_output() -> None: + while True: + if not receiver.thread.is_alive() and receiver.terminal != "end": + raise ConnectionError("input interrupted or rejected") + self.runtime.check_input(access.epoch, synchronizing=True) + item = outbox.take() + if item is not None: + self.runtime.check_input(access.epoch) + raw = encode_reply(access.epoch, *item) + async with asyncio.timeout(self.io_timeout): + await context.write(raw) + del raw + item = None + elif receiver.terminal == "end" and outbox.drained(): + return + else: + await asyncio.sleep(0.01) + + tasks = [asyncio.create_task(read_input()), asyncio.create_task(write_output())] + await asyncio.gather(*tasks) + except (TimeoutError, OSError, ValueError, RuntimeError): + try: + # Even error-status delivery can block behind an unread result. + # It must never delay closing ingress/releasing local buffers. + async with asyncio.timeout(self.io_timeout): + await context.abort( + grpc.StatusCode.UNAVAILABLE, + "stream interrupted; fresh controller grant required", + ) + except TimeoutError: + context.set_code(grpc.StatusCode.UNAVAILABLE) + except Exception as exc: + # A blocked peer can cancel the HTTP/2 call while the server is + # sending its final status. Cleanup still owns the real outcome. + if not isinstance(exc, grpc.RpcError): + raise + finally: + for task in tasks: + task.cancel() + await asyncio.gather(*tasks, return_exceptions=True) + if left is not None: + left.close() + if receiver is not None and receiver.thread.ident is not None: + deadline = time.monotonic() + 1 + while receiver.thread.is_alive() and time.monotonic() < deadline: + await asyncio.sleep(0.01) + elif right is not None: + right.close() + outbox.close() + drained = receiver is None or not receiver.thread.is_alive() + if drained: + self._complete(receiver, reservation, outbox) + else: + self._quarantine_thread = threading.Thread( + target=self._finish_quarantine, + args=(receiver, reservation, outbox), + daemon=True, + name="perception-network-quarantine", + ) + self._quarantine_thread.start() + # A hung trusted consumer is quarantined: no replacement grant or + # premature reservation release while it still owns raw bytes. + + def _finish_quarantine( + self, receiver: StreamingIngress, reservation: Any, outbox: LatestReplies + ) -> None: + receiver.thread.join() # Bounded to this one already-quarantined stream. + self._complete(receiver, reservation, outbox) + + def _complete( + self, receiver: StreamingIngress | None, reservation: Any, outbox: LatestReplies + ) -> None: + if reservation is not None: + reservation.release() + with self.lock: + self.last = { + "ingress": receiver.snapshot() if receiver else None, + "reply_drops": outbox.dropped, + "network_threads_drained": True, + "transport": "grpc-tls-binary/v1", + "actuation_allowed": False, + } + self.active = None + + +class GrpcStreamClient: + """Explicit async source/result API. Exactly one writer and one reader.""" + + def __init__( + self, target: str, roots: bytes, access: StreamAccess, *, timeout: float = 0.25 + ) -> None: + if not roots or not 0 < timeout <= 2: + raise ValueError("trusted TLS roots and bounded timeout required") + self.access, self.timeout = access, timeout + self.channel = grpc.aio.secure_channel( + target, + grpc.ssl_channel_credentials(root_certificates=roots), + options=OPTIONS + + ( + ("grpc.max_send_message_length", INPUT_CHUNK), + ("grpc.max_receive_message_length", MAX_REPLY), + ), + compression=grpc.Compression.NoCompression, + ) + self.call = self.channel.stream_stream(METHOD)( + metadata=access.metadata(), wait_for_ready=False + ) + self.last_reply = -1 + self._sending = self._reading = False + + async def open(self) -> None: + try: + async with asyncio.timeout(2): + metadata = await self.call.initial_metadata() + if dict(metadata).get("x-stream-binding") != wire.binding(self.access.epoch): + raise wire.StreamWireError("server did not admit current input") + await self._send( + ( + wire.open_packet( + self.access.epoch, self.access.session_id, self.access.session_generation + ), + ) + ) + except BaseException: + self.call.cancel() + raise + + async def _send(self, pieces: Iterable[bytes | memoryview]) -> None: + if self._sending: + self.call.cancel() + raise wire.StreamWireError("one source writer per stream") + self._sending = True + try: + async with asyncio.timeout(self.timeout): + for piece in pieces: + for offset in range(0, len(piece), INPUT_CHUNK): + await self.call.write(bytes(piece[offset : offset + INPUT_CHUNK])) + except BaseException: + self.call.cancel() + raise + finally: + self._sending = False + + async def send(self, event: LiveIngressEvent) -> None: + if (event.session_id, event.session_generation) != ( + self.access.session_id, + self.access.session_generation, + ): + self.call.cancel() + raise wire.StreamWireError("source acquisition binding changed") + await self._send( + piece for pair in wire.event_packets(self.access.epoch, event) for piece in pair + ) + + async def end(self) -> None: + await self._send((wire.terminal_packet(self.access.epoch),)) + await self.call.done_writing() + + async def receive(self, *, timeout: float = 2) -> tuple[int, bytes] | None: + if not 0 < timeout <= 30: + raise ValueError("invalid bounded result deadline") + if self._reading: + self.call.cancel() + raise wire.StreamWireError("one result reader per stream") + self._reading = True + try: + async with asyncio.timeout(timeout): + raw = await self.call.read() + if raw is grpc.aio.EOF: + return None + sequence, payload = decode_reply(self.access.epoch, raw) + if sequence <= self.last_reply: + raise wire.StreamWireError("network result sequence regressed") + self.last_reply = sequence + return sequence, payload + except BaseException: + self.call.cancel() + raise + finally: + self._reading = False + + async def close(self) -> None: + self.call.cancel() + await self.channel.close() diff --git a/src/k1link/perception/streaming_network.py b/src/k1link/perception/streaming_network.py new file mode 100644 index 0000000..57e0f84 --- /dev/null +++ b/src/k1link/perception/streaming_network.py @@ -0,0 +1,93 @@ +"""Bounded network envelopes; no model, acquisition or ownership authority. + +Replies wrap existing domain bytes, not a second scene ontology. A complete +reply is at most 1 MiB in this candidate; larger outputs need a separately +reviewed fragmentation adapter, never an automatic enlargement of the queue. +""" + +from __future__ import annotations + +import struct +import threading +from collections import deque +from hashlib import sha256 + +from . import streaming_wire as wire +from .realtime_contract import StreamStart + +INPUT_CHUNK = 64 * 1024 +MAX_RESULT = 1024 * 1024 +REPLY = struct.Struct("!4sQ32s32s") +MAX_REPLY = MAX_RESULT + REPLY.size +# Application-owned bytes, including a reply being serialized/written and two +# pending replies. Native gRPC/HTTP2/socket overhead also needs measured RSS. +BRIDGE_RESERVATION = 4 * MAX_REPLY + 4 * INPUT_CHUNK + + +def encode_reply(epoch: StreamStart, sequence: int, payload: bytes) -> bytes: + wire.decimal(sequence) + if type(payload) is not bytes or not 0 < len(payload) <= MAX_RESULT: + raise wire.StreamWireError("result exceeds bounded binary envelope") + return ( + REPLY.pack(b"MCR3", sequence, bytes.fromhex(wire.binding(epoch)), sha256(payload).digest()) + + payload + ) + + +def decode_reply(epoch: StreamStart, raw: bytes) -> tuple[int, bytes]: + if not REPLY.size < len(raw) <= MAX_REPLY: + raise wire.StreamWireError("invalid reply length") + magic, sequence, binding, digest = REPLY.unpack_from(raw) + payload = raw[REPLY.size :] + if ( + magic != b"MCR3" + or binding.hex() != wire.binding(epoch) + or sha256(payload).digest() != digest + ): + raise wire.StreamWireError("reply identity or integrity mismatch") + return sequence, payload + + +class LatestReplies: + """Two pending immutable results, one consumer, no event-loop callback queue.""" + + def __init__(self) -> None: + self.lock = threading.Lock() + self.pending: deque[tuple[int, bytes]] = deque() + self.last_sequence = -1 + self.dropped = 0 + self.finished = self.closed = False + + def put(self, sequence: int, payload: bytes) -> bool: + wire.decimal(sequence) + if type(payload) is not bytes or not 0 < len(payload) <= MAX_RESULT: + raise wire.StreamWireError("result exceeds bounded binary envelope") + with self.lock: + if self.closed or self.finished: + return False + if sequence <= self.last_sequence: + raise wire.StreamWireError("result sequence regressed") + self.last_sequence = sequence + if len(self.pending) == 2: + self.pending.popleft() + self.dropped += 1 + self.pending.append((sequence, payload)) + return True + + def take(self) -> tuple[int, bytes] | None: + with self.lock: + return self.pending.popleft() if self.pending else None + + def finish(self) -> None: + with self.lock: + self.finished = True + + def drained(self) -> bool: + with self.lock: + return self.finished and not self.pending + + def close(self) -> None: + with self.lock: + self.closed = True + self.dropped += len(self.pending) + self.pending.clear() diff --git a/tests/test_perception_streaming_grpc.py b/tests/test_perception_streaming_grpc.py new file mode 100644 index 0000000..7b901be --- /dev/null +++ b/tests/test_perception_streaming_grpc.py @@ -0,0 +1,463 @@ +"""Small TLS/socket integration checks, not a model benchmark or load test.""" + +# ruff: noqa: E402 -- optional transport extra must be checked before its import. + +import asyncio +import os +import subprocess +import threading +from contextlib import asynccontextmanager +from dataclasses import replace +from hashlib import sha256 + +import pytest + +grpc = pytest.importorskip("grpc", reason="install the perception-stream extra") + +from k1link.compute.live_perception import LiveIngressEvent +from k1link.perception import streaming_wire as wire +from k1link.perception.graph_contracts import GraphState +from k1link.perception.realtime_contract import StreamStart +from k1link.perception.streaming_continuity import ResumeEvidence, StreamSuspended +from k1link.perception.streaming_grpc import GrpcStreamClient, GrpcStreamEndpoint +from k1link.perception.streaming_lifecycle import StreamingLifecycle +from k1link.perception.streaming_network import ( + MAX_RESULT, + LatestReplies, + decode_reply, + encode_reply, +) +from k1link.perception.streaming_queue import StreamMailbox + + +def identity(): + return StreamStart("run", "source", "worker", "epoch", 1, *(["a" * 64] * 4), "clock", "live") + + +def event(sequence=1, payload=b"synthetic-raw-points", stamp=1_000_000_000): + return LiveIngressEvent( + sequence, + "capture", + 1, + "lidar", + "lidar", + sequence, + 1_799_999_999_123_456_789, + stamp, + payload, + ) + + +@pytest.fixture(scope="module") +def tls(tmp_path_factory): + root = tmp_path_factory.mktemp("stream-tls") + key, cert = root / "key.pem", root / "cert.pem" + subprocess.run( + [ + "openssl", + "req", + "-x509", + "-newkey", + "rsa:2048", + "-nodes", + "-days", + "1", + "-subj", + "/CN=localhost", + "-keyout", + str(key), + "-out", + str(cert), + ], + check=True, + capture_output=True, + timeout=10, + ) + return cert.read_bytes(), key.read_bytes() + + +async def eventually(predicate): + async with asyncio.timeout(2): + while not predicate(): + await asyncio.sleep(0.01) + + +@asynccontextmanager +async def harness(tmp_path, tls, *, consume=None, **options): + clock = [1_000_000_000] + run = StreamingLifecycle( + identity(), + tmp_path, + StreamMailbox(), + threading.Event(), + clock_ns=lambda: clock[0], + recover_input=True, + source_clock_ns=lambda: clock[0], + ) + run.ready() + seen = [] + + def accept(value): + seen.append(value) + if consume: + consume(value) + else: + endpoint.publish( + run.continuity.epoch, value.ingress_sequence, sha256(value.payload).digest() + ) + + endpoint = GrpcStreamEndpoint(run, accept, lambda _: None, **options) + server, port = await endpoint.serve("localhost:0", certificate=tls[0], private_key=tls[1]) + clients = [] + + def client(access=None, **kwargs): + access = access or endpoint.issue(run.continuity.epoch, "capture", 1) + result = GrpcStreamClient(f"localhost:{port}", tls[0], access, **kwargs) + clients.append(result) + return result + + try: + yield run, endpoint, client, seen, clock + finally: + for item in clients: + await item.close() + await server.stop(0) + await eventually(lambda: endpoint.active is None) + assert run.close() + assert run.mailbox.bytes == 0 + + +def test_delivers_before_eof_both_directions_and_explicit_result_drain(tmp_path, tls): + async def check(): + async with harness(tmp_path, tls) as (run, endpoint, client, seen, _): + stream = client() + await stream.open() + original = event() + await stream.send(original) + assert await stream.receive() == (1, sha256(original.payload).digest()) + assert seen == [original] + assert not run.mailbox.done + await stream.end() + # Input EOF alone must not discard delayed graph results. + await eventually(lambda: run.mailbox.done) + assert endpoint.active is not None + endpoint.finish_results(identity()) + assert await stream.receive() is None + await eventually(lambda: endpoint.active is None) + assert endpoint.last["ingress"]["terminal"] == "end" + + asyncio.run(check()) + + +@pytest.mark.parametrize("fault", ["token", "binding", "expired", "duplicate"]) +def test_untrusted_grant_cannot_reserve_memory_or_stop_owner(tmp_path, tls, fault): + async def check(): + async with harness(tmp_path, tls) as (run, endpoint, client, _, _): + access = endpoint.issue(identity(), "capture", 1) + if fault == "token": + bad = replace(access, token="0" * 64) + elif fault == "binding": + bad = replace(access, epoch=replace(identity(), calibration_sha256="f" * 64)) + elif fault == "expired": + a, b, _ = endpoint.grant + endpoint.grant = (a, b, 0) + bad = access + else: + with pytest.raises(ValueError): + endpoint._claim((*access.metadata(), *access.metadata())) + bad = replace(access, token="0" * 64) + stream = client(bad) + with pytest.raises((grpc.aio.AioRpcError, wire.StreamWireError)): + await stream.open() + assert endpoint.active is None and run.mailbox.bytes == 0 + assert run.state == GraphState.RUNNING and endpoint.grant is not None + + asyncio.run(check()) + + +def test_disconnect_new_controller_epoch_stale_secret_and_local_lease(tmp_path, tls): + async def check(): + async with harness(tmp_path, tls) as (run, endpoint, client, _, clock): + stream = client() + original_access = stream.access + await stream.open() + await stream.send(event()) + await stream.receive() + with pytest.raises(ValueError): + endpoint.issue(identity(), "capture", 1) + await stream.close() + await eventually(lambda: endpoint.active is None) + assert run.state == GraphState.RUNNING + assert run.continuity.phase == "waiting" and run.lease.start == identity() + clock[0] += 100_000_000 + run.renew(identity()) # Local controller, unrelated to network traffic. + epoch = run.begin_input(identity()) + access = endpoint.issue(epoch, "capture", 1) + stale = client(original_access) + with pytest.raises((grpc.aio.AioRpcError, wire.StreamWireError)): + await stale.open() + assert run.continuity.epoch == epoch and run.state == GraphState.RUNNING + resumed = client(access) + await resumed.open() + # Transport does NOT invent a decoded keyframe/causal sensor proof. + assert run.continuity.phase == "synchronizing" + with pytest.raises(StreamSuspended): + endpoint.publish(epoch, 1, b"premature") + run.resume_input(epoch, ResumeEvidence(*([clock[0]] * 4), True), lambda: None) + await resumed.send(event(2, stamp=clock[0])) + assert (await resumed.receive())[0] == 2 + with pytest.raises(StreamSuspended): + endpoint.publish(identity(), 3, b"obsolete") + await resumed.end() + endpoint.finish_results(epoch) + assert await resumed.receive() is None + + asyncio.run(check()) + + +def test_input_idle_deadline_pauses_without_source_eof(tmp_path, tls): + async def check(): + async with harness(tmp_path, tls, idle_timeout=0.15) as (run, endpoint, client, _, _): + stream = client() + await stream.open() + await stream.send(event()) + await stream.receive() + await eventually(lambda: endpoint.active is None) + assert run.state == GraphState.RUNNING + assert run.continuity.phase == "waiting" and not run.mailbox.done + + asyncio.run(check()) + + +def test_rpc_half_close_without_wire_end_is_not_success(tmp_path, tls): + async def check(): + async with harness(tmp_path, tls) as (run, endpoint, client, _, _): + stream = client() + await stream.open() + await stream.send(event()) + await stream.receive() + await stream.call.done_writing() + with pytest.raises(grpc.aio.AioRpcError): + await stream.receive() + await eventually(lambda: endpoint.active is None) + assert run.continuity.phase == "waiting" and not run.mailbox.done + + asyncio.run(check()) + + +def test_authorized_malformed_wire_remains_fatal(tmp_path, tls): + async def check(): + async with harness(tmp_path, tls) as (run, endpoint, client, _, _): + stream = client() + await stream.open() + await stream.send(event()) + await stream.receive() + await stream.call.write(wire.PREFIX.pack(b"BAD!", wire.OPEN, 1) + b"x") + with pytest.raises(grpc.aio.AioRpcError): + await stream.receive() + await eventually(lambda: endpoint.active is None) + assert run.state == GraphState.STOPPING + + asyncio.run(check()) + + +def test_oversized_rpc_rejected_before_domain_consumer(tmp_path, tls): + async def check(): + async with harness(tmp_path, tls) as (run, endpoint, client, seen, _): + stream = client() + await stream.open() + with pytest.raises(grpc.aio.AioRpcError): + await stream.call.write(b"x" * (64 * 1024 + 1)) + await stream.receive() + assert seen == [] + + asyncio.run(check()) + + +def test_latest_reply_bounds_no_hidden_queue(): + queue = LatestReplies() + for i in range(8): + assert queue.put(i, b"result") + assert len(queue.pending) == 2 and queue.dropped == 6 + assert queue.take() == (6, b"result") + assert queue.take() == (7, b"result") + with pytest.raises(ValueError): + queue.put(7, b"regressed") + queue.finish() + assert not queue.put(8, b"late") and queue.drained() + + +@pytest.mark.parametrize("payload", [b"", b"x" * (MAX_RESULT + 1), bytearray(b"x")]) +def test_result_size_type_rejected_before_queue(payload): + with pytest.raises(ValueError): + LatestReplies().put(0, payload) + + +def test_reply_digest_epoch_and_exact_uint64(): + raw = encode_reply(identity(), (1 << 53) + 1, b"existing-domain-wire") + assert decode_reply(identity(), raw) == ((1 << 53) + 1, b"existing-domain-wire") + for altered in (raw[:-1] + b"?", raw[:20]): + with pytest.raises(ValueError): + decode_reply(identity(), altered) + with pytest.raises(ValueError): + decode_reply(replace(identity(), epoch_id="another"), raw) + + +def test_tls_name_verification_is_not_disabled(tmp_path, tls): + async def check(): + async with harness(tmp_path, tls) as (run, endpoint, _, seen, _): + access = endpoint.issue(identity(), "capture", 1) + server, port = await endpoint.serve( + "127.0.0.1:0", certificate=tls[0], private_key=tls[1] + ) + stream = GrpcStreamClient(f"127.0.0.1:{port}", tls[0], access) + try: + # The certificate is for localhost, not this numeric address. + with pytest.raises((grpc.aio.AioRpcError, wire.StreamWireError)): + await stream.open() + assert endpoint.active is None and not seen and run.mailbox.bytes == 0 + finally: + await stream.close() + await server.stop(0) + + asyncio.run(check()) + + +def test_slow_result_sink_times_out_without_killing_resident_runtime(tmp_path): + class Aborted(Exception): + pass + + async def check(): + run = StreamingLifecycle( + identity(), + tmp_path, + StreamMailbox(), + threading.Event(), + recover_input=True, + source_clock_ns=lambda: 1_000_000_000, + ) + run.ready() + endpoint = GrpcStreamEndpoint( + run, + lambda e: endpoint.publish(identity(), e.ingress_sequence, b"result"), + lambda _: None, + io_timeout=0.05, + ) + access = endpoint.issue(identity(), "capture", 1) + packets = [wire.open_packet(identity(), "capture", 1)] + packets += [ + bytes(piece) for pair in wire.event_packets(identity(), event()) for piece in pair + ] + + class Context: + def auth_context(self): + return {"transport_security_type": (b"ssl",)} + + def invocation_metadata(self): + return access.metadata() + + async def send_initial_metadata(self, _): + pass + + async def read(self): + if packets: + return packets.pop(0) + await asyncio.Event().wait() + + async def write(self, _): + await asyncio.Event().wait() + + async def abort(self, *_): + raise Aborted() + + try: + with pytest.raises(Aborted): + async with asyncio.timeout(1): + await endpoint.exchange(None, Context()) + assert run.state == GraphState.RUNNING and run.continuity.phase == "waiting" + assert endpoint.active is None and run.mailbox.bytes == 0 + finally: + assert run.close() + + asyncio.run(check()) + + +def test_slow_trusted_callback_quarantines_then_releases_without_replacing_owner(tmp_path, tls): + entered, leave = threading.Event(), threading.Event() + + def consume(_): + entered.set() + leave.wait(3) + + async def check(): + async with harness(tmp_path, tls, consume=consume) as (run, endpoint, client, _, _): + stream = client() + try: + await stream.open() + await stream.send(event()) + await eventually(entered.is_set) + await stream.close() + await eventually(lambda: endpoint._quarantine_thread is not None) + assert endpoint.active is not None and run.mailbox.bytes > 0 + with pytest.raises((ValueError, StreamSuspended)): + endpoint.issue(identity(), "capture", 1) + finally: + leave.set() + await eventually(lambda: endpoint.active is None) + assert run.state == GraphState.RUNNING and run.mailbox.bytes == 0 + await eventually(lambda: not endpoint._quarantine_thread.is_alive()) + + asyncio.run(check()) + + +@pytest.mark.skipif( + os.environ.get("NDC_STREAM_WORKER_PROBE") != "1", reason="Worker-only bounded probe" +) +def test_worker_maximum_result_crosses_tls_intact(tmp_path, tls): + payload = b"r" * MAX_RESULT + + async def check(): + async with harness( + tmp_path, + tls, + consume=lambda e: endpoint.publish(identity(), e.ingress_sequence, payload), + ) as (_, endpoint, client, _, _): + stream = client() + await stream.open() + await stream.send(event()) + assert await stream.receive() == (1, payload) + await stream.end() + endpoint.finish_results(identity()) + assert await stream.receive() is None + + asyncio.run(check()) + + +@pytest.mark.skipif( + os.environ.get("NDC_STREAM_WORKER_PROBE") != "1", reason="Worker-only bounded probe" +) +def test_worker_actual_grpc_slow_reader_releases_stream_not_runtime(tmp_path, tls): + payload = b"r" * MAX_RESULT + + async def check(): + async with harness( + tmp_path, + tls, + consume=lambda e: endpoint.publish(identity(), e.ingress_sequence, payload), + io_timeout=0.1, + ) as (run, endpoint, client, _, _): + stream = client() + await stream.open() + # At most 24 tiny inputs; server outputs at most 24 x 1 MiB on Worker. + # Never read results: real gRPC/TLS flow control must reach its bound. + for sequence in range(1, 25): + try: + await stream.send(event(sequence)) + except grpc.aio.AioRpcError: + break + await asyncio.sleep(0.02) + await eventually(lambda: endpoint.active is None) + assert run.state == GraphState.RUNNING and run.continuity.phase == "waiting" + assert run.mailbox.bytes == 0 and endpoint.last["reply_drops"] > 0 + + asyncio.run(check()) diff --git a/uv.lock b/uv.lock index aea6949..8dbd495 100644 --- a/uv.lock +++ b/uv.lock @@ -146,6 +146,27 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/c7/e1/8ccb4a985c5baf82947e48cff18483c2125ea11e55e8a28950730ab6c065/foxglove_sdk-0.25.3-cp310-abi3-win_amd64.whl", hash = "sha256:ac881ae307ba432766e6141d9098ced2059838226d225fbeb20de1c57d782a9f", size = 16496466, upload-time = "2026-06-25T00:24:24.906Z" }, ] +[[package]] +name = "grpcio" +version = "1.83.1" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/e2/b1/46539f5050d7c316a13396d185451f95084a74ddc68b12d818595bef0377/grpcio-1.83.1.tar.gz", hash = "sha256:9cee6fcbf2eb57c4b49451787bfa87be8efc1ca02a0b327dd4b54d44502e362b", size = 13445033, upload-time = "2026-08-28T07:09:11.464Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/85/9e/a3ba13e08bbee5bf6e57597dfe4823961fd7e94c0b8afe3a4bb7dca639f3/grpcio-1.83.1-cp312-cp312-linux_armv7l.whl", hash = "sha256:5acd14c6ddf047de62cbf8745b11103ea91abbf57d1b8edd5395ccd9fcd13abb", size = 6303170, upload-time = "2026-08-28T07:08:08.188Z" }, + { url = "https://files.pythonhosted.org/packages/1c/ae/65ce56a2527faa17d02cba4c2231c74047ad898be339486ba87f093bfb66/grpcio-1.83.1-cp312-cp312-macosx_11_0_universal2.whl", hash = "sha256:16138031a47b771860a16a975b53087f4fd5bbdbb2c03a188c5d90ad65d2bdae", size = 12165806, upload-time = "2026-08-28T07:08:10.309Z" }, + { url = "https://files.pythonhosted.org/packages/4e/91/40432480088a2243d360864de072ed5b78c4ebbaabd29c28918f1e1b1454/grpcio-1.83.1-cp312-cp312-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:5ccc26715fd4defca5e129e280dd883b1737b65045ec50ffe22ce42104089519", size = 6872490, upload-time = "2026-08-28T07:08:12.355Z" }, + { url = "https://files.pythonhosted.org/packages/c8/62/3da2300c8c79fd20a78a8a4bb6251e5068d9af33bc8fd389b98fec35e8a3/grpcio-1.83.1-cp312-cp312-manylinux2014_i686.manylinux_2_17_i686.whl", hash = "sha256:b74f2a1d9ab1dfa3e263ef33d581613679b78d0884babf11671af26e45570ead", size = 7618367, upload-time = "2026-08-28T07:08:14.025Z" }, + { url = "https://files.pythonhosted.org/packages/bc/19/9fc702e31a631262d7a752fa699f6022821e707fefc8bff49b1550a57729/grpcio-1.83.1-cp312-cp312-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:72578aa07a4008f17521ef52debcc3acfd1e2c5426243bc3ffb56a38bfe610b7", size = 7040936, upload-time = "2026-08-28T07:08:15.963Z" }, + { url = "https://files.pythonhosted.org/packages/ec/56/95933cc44cba2429765fa065c951dd529e5771b119d9d2481b4646f1d6a5/grpcio-1.83.1-cp312-cp312-musllinux_1_2_aarch64.whl", hash = "sha256:c12e1fc59c6dc26d10d9144453ddc6cbfe4cd4c31e874ed2d0132f88e685eb8b", size = 7573096, upload-time = "2026-08-28T07:08:17.729Z" }, + { url = "https://files.pythonhosted.org/packages/ac/80/af63359da06b016de48cb111f144703a10043850dafa43ae0a038907b9e8/grpcio-1.83.1-cp312-cp312-musllinux_1_2_i686.whl", hash = "sha256:4910b62f7d12197160bfb7de06d876d64dd12d43483e8292f98f49ca09b628d9", size = 8609442, upload-time = "2026-08-28T07:08:19.777Z" }, + { url = "https://files.pythonhosted.org/packages/3d/fa/f0586c56bdfb8a7a2adda01e0ac2413447cde3141ab09411a5d5afdcffd3/grpcio-1.83.1-cp312-cp312-musllinux_1_2_x86_64.whl", hash = "sha256:9e703effe3ae779925c82ac24fdb82cf4105e1096810151ed9501c5f34546b9c", size = 7984321, upload-time = "2026-08-28T07:08:22.114Z" }, + { url = "https://files.pythonhosted.org/packages/25/8a/14ec05669f9eb295801e26c2ea8c561a1b786b0e3557c2c22131165ab010/grpcio-1.83.1-cp312-cp312-win32.whl", hash = "sha256:a2aea8bd6e0a34f12cbaddb7bb70bec836818789fa5c7ab7572c6b745396a2d4", size = 4395604, upload-time = "2026-08-28T07:08:24.08Z" }, + { url = "https://files.pythonhosted.org/packages/e9/37/8c2f7cc16089e36a3fbacaacc7a3d043912aa0d2dfae5556f6450414ea6e/grpcio-1.83.1-cp312-cp312-win_amd64.whl", hash = "sha256:583bf2e8255040a4a312f9572dfe62a05271437b149550e1a536d5c47d2d1e8a", size = 5161512, upload-time = "2026-08-28T07:08:25.81Z" }, +] + [[package]] name = "h11" version = "0.16.0" @@ -336,6 +357,11 @@ dependencies = [ { name = "uvicorn", extra = ["standard"] }, ] +[package.optional-dependencies] +perception-stream = [ + { name = "grpcio" }, +] + [package.dev-dependencies] dev = [ { name = "mypy" }, @@ -349,6 +375,7 @@ requires-dist = [ { name = "bleak", specifier = "==3.0.2" }, { name = "fastapi", specifier = ">=0.116,<1" }, { name = "foxglove-sdk", specifier = "==0.25.3" }, + { name = "grpcio", marker = "extra == 'perception-stream'", specifier = ">=1.76,<2" }, { name = "httpx", specifier = ">=0.28,<1" }, { name = "lz4", specifier = ">=4.4,<5" }, { name = "missioncore-plugin-sdk", editable = "packages/plugin-sdk" }, @@ -360,6 +387,7 @@ requires-dist = [ { name = "typer", specifier = ">=0.15,<1" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.35,<1" }, ] +provides-extras = ["perception-stream"] [package.metadata.requires-dev] dev = [