From da60feff90231eef2a61c9ae074ff79ae417a5f3 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 12:21:25 +0300 Subject: [PATCH] feat(perception): add bounded binary stream ingress --- .../pilot_stream_ingress_probe.py | 312 +++++++++++++++++ src/k1link/perception/streaming_ingress.py | 293 ++++++++++++++++ src/k1link/perception/streaming_queue.py | 43 ++- src/k1link/perception/streaming_sender.py | 91 +++++ src/k1link/perception/streaming_wire.py | 211 ++++++++++++ tests/test_perception_streaming_ingress.py | 313 ++++++++++++++++++ 6 files changed, 1262 insertions(+), 1 deletion(-) create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py create mode 100644 src/k1link/perception/streaming_ingress.py create mode 100644 src/k1link/perception/streaming_sender.py create mode 100644 src/k1link/perception/streaming_wire.py create mode 100644 tests/test_perception_streaming_ingress.py diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py new file mode 100644 index 0000000..ddd7fdb --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_stream_ingress_probe.py @@ -0,0 +1,312 @@ +"""CPU-only, bounded IPC evidence on Worker; NOT a model/network qualification. + +Recorded mode transfers original fMP4 init/media plus existing normalized +point/pose increments at original 1x release times. Receiver is never told the +source length. The consumer only hashes raw input, it does not decode/infer. +Truncated mode is explicitly synthetic and exercises multi-fragment failure. +""" + +import argparse +import hashlib +import json +import resource +import socket +import struct +import subprocess +import sys +import threading +import time +from datetime import UTC, datetime +from pathlib import Path + +from pilot_source import SensorArchive, camera_events, merged_events + +from k1link.compute.live_perception import LiveIngressEvent +from k1link.perception import streaming_wire as wire +from k1link.perception.realtime_contract import StreamStart +from k1link.perception.streaming_ingress import StreamingIngress +from k1link.perception.streaming_lifecycle import StreamingLifecycle +from k1link.perception.streaming_queue import StreamMailbox +from k1link.perception.streaming_sender import StreamingSender + + +def record(event): + return { + **wire.event_metadata(event), + "payload_bytes": len(event.payload), + "payload_sha256": hashlib.sha256(event.payload).hexdigest(), + } + + +def emit_json(stream, value): + stream.write(json.dumps(value, sort_keys=True) + "\n") + stream.flush() + + +def bounded_file(path, maximum): + with path.open("rb") as stream: + raw = stream.read(maximum + 1) + if not 0 < len(raw) <= maximum: + raise ValueError("source member exceeds bound") + return raw + + +def recorded_events(root): + camera_root = Path("/camera") + camera_index = camera_root / "index.jsonl" + first = next(camera_events(camera_index, 1)).time_ns + zero = first - 500_000_000 + wall = time.monotonic_ns() + 50_000_000 + seq = 1 + yield ( + LiveIngressEvent( + seq, + "recorded-acquisition", + 1, + "camera-init", + "sensor.camera.right", + 0, + 0, + zero, + bounded_file(camera_root / "init.mp4", wire.MAX_FRAGMENT), + ), + wall, + ) + archive = SensorArchive(Path("/sensor-source.npz")) + try: + for observation in merged_events(archive, camera_index, 32): + if observation.time_ns < zero: + continue + due = wall + observation.time_ns - zero + threading.Event().wait(max(0, (due - time.monotonic_ns()) / 1e9)) + seq += 1 + utc = 0 # The normalized sensor archive has no UTC evidence; do not invent it. + if observation.channel == "camera": + row = observation.value + path = (camera_root / row["path"]).resolve() + if not path.is_relative_to(camera_root): + raise ValueError("camera member escaped source root") + raw = bounded_file(path, wire.MAX_FRAGMENT) + if len(raw) != row["length"] or hashlib.sha256(raw).hexdigest() != row["sha256"]: + raise ValueError("camera chunk integrity failed") + modality, source_id, utc = ( + "camera-frame", + "sensor.camera.right", + row["host_epoch_ns"], + ) + elif observation.channel == "points": + xyz, intensity = observation.value + raw = struct.pack(" None: + if not 0 < fragment_timeout <= 2 or not 0 < idle_timeout <= 30: + raise ValueError("invalid bounded ingress timeouts") + wire.identifier(session_id) + wire.decimal(session_generation) + if session_generation < 1: + raise ValueError("invalid acquisition generation") + self.connection, self.runtime = connection, runtime + connection.setblocking(False) + self.session_id, self.session_generation = session_id, session_generation + self.consume, self.notice = consume, notice + self.fragment_timeout, self.idle_timeout = fragment_timeout, idle_timeout + self.binding = wire.binding(runtime.start) + self.opened = False + self.error: str | None = None + self.terminal: str | None = None + self.counts: Counter[str] = Counter() + self.last_ingress_sequence = 0 + self.channel_sequences: dict[str, tuple[str, int, int]] = {} + self.camera_initialized = False + self.thread = threading.Thread( + target=self._serve, name="perception-binary-ingress", daemon=True + ) + runtime.track_thread(self.thread) + + def start(self) -> None: + self.runtime.check_current(self.runtime.start) + self.thread.start() + + def join(self, timeout: float = 1.0) -> bool: + self.thread.join(timeout=timeout) + return not self.thread.is_alive() + + def _check(self, deadline: float) -> None: + self.runtime.check_current(self.runtime.start) + if time.monotonic() >= deadline: + raise wire.StreamWireError("ingress deadline exceeded") + + def _read_into(self, target: memoryview, deadline: float) -> None: + offset = 0 + while offset < len(target): + self._check(deadline) + if not select.select( + [self.connection], [], [], min(0.05, max(0, deadline - time.monotonic())) + )[0]: + continue + try: + read = self.connection.recv_into(target[offset:]) + except BlockingIOError: + continue + if read == 0: + raise wire.StreamWireError("unexpected EOF without explicit End") + self.counts["wire_bytes"] += read + offset += read + self._check(deadline) + + def _header(self, deadline: float) -> tuple[int, dict[str, Any]]: + prefix = bytearray(wire.PREFIX.size) + self._read_into(memoryview(prefix), deadline) + magic, kind, size = wire.PREFIX.unpack(prefix) + if ( + magic != wire.MAGIC + or kind not in (wire.OPEN, wire.FRAGMENT, wire.END, wire.CANCEL, wire.GAP) + or not 0 < size <= wire.MAX_HEADER + ): + raise wire.StreamWireError("invalid frame prefix or header size") + raw = bytearray(size) + self._read_into(memoryview(raw), deadline) + return kind, wire.parse_header(raw) + + def _bound(self, header: dict[str, Any]) -> None: + if header.get("binding") != self.binding: + raise wire.StreamWireError("frame belongs to another StreamStart") + self.runtime.check_current(self.runtime.start) + + def _fragment(self, header: dict[str, Any]) -> tuple[dict[str, Any], int, int, int]: + if set(header) != { + "binding", + "event", + "total_bytes", + "offset", + "fragment_bytes", + "payload_sha256", + "fragment_sha256", + }: + raise wire.StreamWireError("fragment fields changed") + self._bound(header) + event = wire.validate_event_metadata(header["event"]) + if ( + event["session_id"] != self.session_id + or event["session_generation"] != self.session_generation + ): + raise wire.StreamWireError("event acquisition binding changed") + total, offset, size = (header[x] for x in ("total_bytes", "offset", "fragment_bytes")) + if any(type(x) is not int for x in (total, offset, size)) or not ( + 0 < total <= wire.PAYLOAD_LIMITS[event["modality"]] + and 0 <= offset < total + and 0 < size <= wire.MAX_FRAGMENT + and size == min(wire.MAX_FRAGMENT, total - offset) + ): + raise wire.StreamWireError("invalid fragment length or offset") + wire.digest(header["payload_sha256"]) + wire.digest(header["fragment_sha256"]) + return event, total, offset, size + + def _observation(self, first: dict[str, Any], deadline: float) -> None: + event, total, offset, size = self._fragment(first) + modality = event["modality"] + if offset != 0 or event["ingress_sequence"] <= self.last_ingress_sequence: + raise wire.StreamWireError("duplicate, out-of-order or incomplete observation") + previous = self.channel_sequences.get(modality) + if previous and ( + event["source_id"] != previous[0] + or event["source_sequence"] <= previous[1] + or event["received_monotonic_ns"] < previous[2] + ): + raise wire.StreamWireError("channel identity, sequence or source clock regressed") + if modality == "camera-frame" and not self.camera_initialized: + raise wire.StreamWireError("camera frame requires new codec initialization") + if modality == "camera-frame" and previous and event["source_sequence"] != previous[1] + 1: + raise wire.StreamWireError("camera sequence gap requires new codec initialization") + self.counts["observations_started"] += 1 + reservation = self.runtime.mailbox.reserve_ingress(2 * total) + raw = None + admitted = None + try: + raw = bytearray(total) + header = first + while True: + chunk = memoryview(raw)[offset : offset + size] + try: + self._read_into(chunk, deadline) + if sha256(chunk).hexdigest() != header["fragment_sha256"]: + raise wire.StreamWireError("fragment integrity mismatch") + finally: + chunk.release() + self.counts["fragments"] += 1 + offset += size + if offset == total: + break + kind, header = self._header(deadline) + if kind != wire.FRAGMENT: + raise wire.StreamWireError("incomplete observation before control event") + next_event, next_total, next_offset, size = self._fragment(header) + if ( + next_event != event + or next_total != total + or next_offset != offset + or header["payload_sha256"] != first["payload_sha256"] + ): + raise wire.StreamWireError("interleaved or discontinuous fragments") + if sha256(raw).hexdigest() != first["payload_sha256"]: + raise wire.StreamWireError("observation integrity mismatch") + self._check(deadline) + admitted = LiveIngressEvent(**event, payload=bytes(raw)) + raw = None # Immutable callback bytes remain fully reserved. + self.consume(admitted) + self._check(deadline) + self.counts["ingress_sequence_gaps"] += ( + event["ingress_sequence"] - self.last_ingress_sequence - 1 + ) + self.last_ingress_sequence = event["ingress_sequence"] + self.channel_sequences[modality] = ( + event["source_id"], + event["source_sequence"], + event["received_monotonic_ns"], + ) + if modality == "camera-init": + self.camera_initialized = True + self.channel_sequences.pop("camera-frame", None) + self.counts["observations_completed"] += 1 + finally: + admitted = raw = None + reservation.release() + + def _gap(self, header: dict[str, Any]) -> None: + if set(header) != {"binding", "modality", "reason", "count"}: + raise wire.StreamWireError("gap fields changed") + if header["modality"] not in wire.PAYLOAD_LIMITS or header["reason"] not in ( + "unavailable", + "source-gap", + "overload", + ): + raise wire.StreamWireError("invalid gap notice") + self.counts["declared_gap_observations"] += wire.uint64(header["count"]) + if header["modality"].startswith("camera"): + self.camera_initialized = False + self.notice(dict(header)) + self.runtime.check_current(self.runtime.start) + self.counts["gap_notices"] += 1 + + def _serve(self) -> None: + reservation = None + try: + # Covers raw header + JSON decoding copies; Python object overhead + # belongs to the separately measured RSS envelope, not payload bytes. + reservation = self.runtime.mailbox.reserve_ingress( + 3 * wire.MAX_HEADER + wire.PREFIX.size + ) + kind, header = self._header(time.monotonic() + self.idle_timeout) + expected = wire.parse_header( + bytearray( + wire.open_packet( + self.runtime.start, + self.session_id, + self.session_generation, + )[wire.PREFIX.size :] + ) + ) + if kind != wire.OPEN or header != expected: + raise wire.StreamWireError("StreamStart or acquisition handshake mismatch") + self.opened = True + while True: + kind, header = self._header(time.monotonic() + self.idle_timeout) + self._bound(header) + if kind == wire.FRAGMENT: + self._observation(header, time.monotonic() + self.fragment_timeout) + elif kind == wire.GAP: + self._gap(header) + elif kind in (wire.END, wire.CANCEL) and set(header) == {"binding"}: + self.terminal = "cancelled" if kind == wire.CANCEL else "end" + if kind == wire.CANCEL: + self.runtime.request_stop("cancelled") + else: + self.runtime.mailbox.finish() + return + else: + raise wire.StreamWireError("unexpected stream control event") + except Exception as exc: + self.error = str(exc) + self.terminal = "failed" + if self.opened: + self.runtime.mailbox.finish(self.error) + self.runtime.request_stop("failed") + finally: + self.connection.close() + if reservation is not None: + reservation.release() + + def snapshot(self) -> dict[str, Any]: + return { + "schema_version": wire.SCHEMA, + "opened": self.opened, + "terminal": self.terminal, + "error": self.error, + "counts": dict(self.counts), + "incomplete_observations": self.counts["observations_started"] + - self.counts["observations_completed"], + "source_duration_known": False, + "requires_eof_before_delivery": False, + "transport": "controller-supplied-ipc-socket", + "network_qualified": False, + "actuation_allowed": False, + } diff --git a/src/k1link/perception/streaming_queue.py b/src/k1link/perception/streaming_queue.py index 1f72f83..c725e9e 100644 --- a/src/k1link/perception/streaming_queue.py +++ b/src/k1link/perception/streaming_queue.py @@ -17,6 +17,21 @@ StreamBundle = Mapping[str, Any] DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed")) +class IngressReservation: + """Raw/reassembly bytes retained outside the decoded work queue. + + Trusted adapters reserve BEFORE allocation and release AFTER borrowed raw + buffers/callbacks are gone. Closing ingress never frees a live reservation. + This is byte ownership, not another pending camera slot. + """ + + def __init__(self, mailbox: StreamMailbox, size: int) -> None: + self._mailbox, self.size = mailbox, size + + def release(self) -> None: + self._mailbox._release_ingress(self) + + class StreamMailbox: def __init__(self, capacity: int = 2, byte_limit: int = 16 * 1024 * 1024) -> None: if type(capacity) is not int or not 1 <= capacity <= 2: @@ -31,6 +46,7 @@ class StreamMailbox: self._recent_drops: deque[dict[str, Any]] = deque(maxlen=256) self._owned: dict[int, tuple[StreamBundle, int]] = {} self._active: set[int] = set() + self._ingress: dict[IngressReservation, int] = {} self._last_sequence = -1 self.done = False self.error: str | None = None @@ -49,7 +65,32 @@ class StreamMailbox: @property def quiescent(self) -> bool: with self.condition: - return self.done and not self._owned and not self.external_pending + return self.done and not self._owned and not self.external_pending and not self._ingress + + def reserve_ingress(self, size: int) -> IngressReservation: + """Charge raw buffers to the SAME budget as queued and active inputs. + + Fail closed on exhaustion; do not wait and silently stretch source time. + Decoder scratch must also be reserved before allocation; retained decoded + work uses the existing bundle ownership. No adapter may hide a buffer. + """ + if type(size) is not int or size <= 0: + raise ValueError("invalid ingress reservation") + with self.condition: + if self.done or self.bytes + size > self.byte_limit or len(self._ingress) >= 8: + raise ValueError("ingress byte budget exhausted or closed") + reservation = IngressReservation(self, size) + self._ingress[reservation] = size + self.bytes += size + self.peak_bytes = max(self.peak_bytes, self.bytes) + return reservation + + def _release_ingress(self, reservation: IngressReservation) -> None: + with self.condition: + if reservation not in self._ingress: + raise ValueError("ingress release ownership mismatch") + self.bytes -= self._ingress.pop(reservation) + self.condition.notify_all() def _drop(self, bundle: StreamBundle, reason: str) -> None: self.drop_counts[reason] += 1 diff --git a/src/k1link/perception/streaming_sender.py b/src/k1link/perception/streaming_sender.py new file mode 100644 index 0000000..a58e910 --- /dev/null +++ b/src/k1link/perception/streaming_sender.py @@ -0,0 +1,91 @@ +"""Synchronous, deadline-bounded IPC writer for existing raw ingress events. + +The caller owns pacing and already-committed source bytes. No background queue, +recording inventory or retry is created; timeout is an explicit failed source +outcome. A new connection requires a new controller-approved StreamStart. +""" + +from __future__ import annotations + +import select +import socket +import time +from collections.abc import Callable, Iterable + +from k1link.compute.live_perception import LiveIngressEvent + +from . import streaming_wire as wire +from .realtime_contract import StreamStart + + +class StreamingSender: + def __init__( + self, + connection: socket.socket, + start: StreamStart, + session_id: str, + session_generation: int, + check_current: Callable[[], None], + *, + timeout: float = 0.25, + ) -> None: + if not 0 < timeout <= 2: + raise ValueError("invalid bounded send timeout") + self.connection, self.identity, self.check_current = connection, start, check_current + self.session_id, self.session_generation = session_id, session_generation + self.timeout, self.closed = timeout, False + connection.setblocking(False) + try: + self._send((wire.open_packet(start, session_id, session_generation),)) + except Exception: + self.close() + raise + + def _send(self, pieces: Iterable[bytes | memoryview]) -> None: + if self.closed: + raise wire.StreamWireError("sender is closed") + deadline = time.monotonic() + self.timeout + try: + for piece in pieces: + view = memoryview(piece) + offset = 0 + while offset < len(view): + self.check_current() + remaining = deadline - time.monotonic() + if remaining <= 0: + raise wire.StreamWireError("send deadline exceeded") + if not select.select([], [self.connection], [], min(0.05, remaining))[1]: + continue + try: + count = self.connection.send(view[offset:]) + except BlockingIOError: + continue + if count == 0: + raise wire.StreamWireError("source connection closed") + offset += count + self.check_current() + if time.monotonic() >= deadline: + raise wire.StreamWireError("send deadline exceeded") + except Exception: + self.close() + raise + + def send(self, event: LiveIngressEvent) -> None: + if ( + event.session_id != self.session_id + or event.session_generation != self.session_generation + ): + self.close() + raise wire.StreamWireError("source acquisition binding changed") + self._send(piece for pair in wire.event_packets(self.identity, event) for piece in pair) + + def gap(self, *, modality: str, reason: str, count: int) -> None: + self._send((wire.gap_packet(self.identity, modality=modality, reason=reason, count=count),)) + + def end(self, *, cancel: bool = False) -> None: + self._send((wire.terminal_packet(self.identity, cancel=cancel),)) + self.close() + + def close(self) -> None: + self.closed = True + self.connection.close() diff --git a/src/k1link/perception/streaming_wire.py b/src/k1link/perception/streaming_wire.py new file mode 100644 index 0000000..d246a1b --- /dev/null +++ b/src/k1link/perception/streaming_wire.py @@ -0,0 +1,211 @@ +"""Bounded v2 mapping of existing raw-first LiveIngressEvent semantics. + +This IPC candidate carries binary payloads, not base64/PNG/archive uploads. +It is not the selected/authenticated network transport. A trusted controller +binds one acquisition session to StreamStart; hashes provide integrity, NOT +authentication. No peer-supplied paths, commands, models or calibration loading. +""" + +from __future__ import annotations + +import json +import re +import struct +from collections.abc import Iterator +from hashlib import sha256 +from typing import Any, Final + +from k1link.compute.live_perception import LiveIngressEvent + +from .realtime_contract import StreamStart + +SCHEMA: Final = "missioncore.live-perception-wire/v2" +MAGIC: Final = b"MCI2" +PREFIX: Final = struct.Struct("!4sBI") +OPEN, FRAGMENT, END, CANCEL, GAP = range(1, 6) +MAX_HEADER: Final = 64 * 1024 +MAX_FRAGMENT: Final = 1024 * 1024 +PAYLOAD_LIMITS: Final = { + "camera-init": MAX_FRAGMENT, + "camera-frame": MAX_FRAGMENT, + "lidar": 2 * MAX_FRAGMENT, + "pose": 2 * MAX_FRAGMENT, +} +EVENT_FIELDS: Final = frozenset( + ( + "ingress_sequence", + "session_id", + "session_generation", + "modality", + "source_id", + "source_sequence", + "captured_at_epoch_ns", + "received_monotonic_ns", + ) +) +INTEGER_FIELDS: Final = EVENT_FIELDS - {"session_id", "modality", "source_id"} +_DECIMAL: Final = re.compile(r"^(0|[1-9][0-9]{0,19})$") +_DIGEST: Final = re.compile(r"^[a-f0-9]{64}$") + + +class StreamWireError(ValueError): + """Malformed, stale, oversized or incomplete transport input.""" + + +def uint64(value: object) -> int: + if not isinstance(value, str) or not _DECIMAL.fullmatch(value): + raise StreamWireError("wire uint64 must be a canonical decimal string") + result = int(value) + if result > (1 << 64) - 1: + raise StreamWireError("wire uint64 overflow") + return result + + +def decimal(value: object) -> str: + if type(value) is not int or not 0 <= value < 1 << 64: + raise StreamWireError("invalid uint64") + return str(value) + + +def identifier(value: object) -> str: + if not isinstance(value, str) or not 1 <= len(value) <= 160: + raise StreamWireError("invalid bounded source identity") + if not value.isascii() or any(ord(c) < 33 or ord(c) > 126 for c in value): + raise StreamWireError("invalid bounded source identity") + return value + + +def digest(value: object) -> str: + if not isinstance(value, str) or not _DIGEST.fullmatch(value): + raise StreamWireError("invalid payload digest") + return value + + +def canonical(value: dict[str, Any]) -> bytes: + return json.dumps(value, sort_keys=True, separators=(",", ":"), allow_nan=False).encode() + + +def start_document(start: StreamStart) -> dict[str, Any]: + result = start.to_dict() + result["lease_generation"] = decimal(start.lease_generation) + return result + + +def binding(start: StreamStart) -> str: + return sha256(canonical(start_document(start))).hexdigest() + + +def _unique_pairs(pairs: list[tuple[str, Any]]) -> dict[str, Any]: + result: dict[str, Any] = {} + for key, value in pairs: + if key in result: + raise StreamWireError("duplicate metadata field") + result[key] = value + return result + + +def parse_header(raw: bytearray) -> dict[str, Any]: + try: + value = json.loads(raw, object_pairs_hook=_unique_pairs) + except (ValueError, UnicodeError, RecursionError) as exc: + raise StreamWireError("invalid bounded metadata") from exc + if not isinstance(value, dict): + raise StreamWireError("metadata must be an object") + return value + + +def packet(kind: int, header: dict[str, Any]) -> bytes: + raw = canonical(header) + if kind not in (OPEN, FRAGMENT, END, CANCEL, GAP) or not 0 < len(raw) <= MAX_HEADER: + raise StreamWireError("invalid frame header") + return PREFIX.pack(MAGIC, kind, len(raw)) + raw + + +def open_packet(start: StreamStart, session_id: str, session_generation: int) -> bytes: + return packet( + OPEN, + { + "schema_version": SCHEMA, + "start": start_document(start), + "session_id": identifier(session_id), + "session_generation": decimal(session_generation), + }, + ) + + +def event_metadata(event: LiveIngressEvent) -> dict[str, Any]: + result = {name: getattr(event, name) for name in EVENT_FIELDS} + for name in INTEGER_FIELDS: + result[name] = decimal(result[name]) + validate_event_metadata(result) + return result + + +def validate_event_metadata(value: object) -> dict[str, Any]: + if not isinstance(value, dict) or set(value) != EVENT_FIELDS: + raise StreamWireError("event fields changed") + result = dict(value) + for name in INTEGER_FIELDS: + result[name] = uint64(value[name]) + for name in ("session_id", "source_id"): + identifier(value[name]) + if ( + not isinstance(value["modality"], str) + or value["modality"] not in PAYLOAD_LIMITS + or result["ingress_sequence"] < 1 + or result["session_generation"] < 1 + ): + raise StreamWireError("invalid raw modality or generation") + return result + + +def event_packets( + start: StreamStart, + event: LiveIngressEvent, +) -> Iterator[tuple[bytes, memoryview]]: + """Iterate one already committed raw event; never assemble a route/archive. + + Source adapter owns the original event bytes. Returned views borrow them; + sender must finish each fragment before advancing this iterator. + """ + metadata = event_metadata(event) + total = len(event.payload) + if not 0 < total <= PAYLOAD_LIMITS[event.modality]: + raise StreamWireError("raw event exceeds modality bound") + whole_hash = sha256(event.payload).hexdigest() + view = memoryview(event.payload) + for offset in range(0, total, MAX_FRAGMENT): + chunk = view[offset : offset + MAX_FRAGMENT] + yield ( + packet( + FRAGMENT, + { + "binding": binding(start), + "event": metadata, + "total_bytes": total, + "offset": offset, + "fragment_bytes": len(chunk), + "payload_sha256": whole_hash, + "fragment_sha256": sha256(chunk).hexdigest(), + }, + ), + chunk, + ) + + +def terminal_packet(start: StreamStart, *, cancel: bool = False) -> bytes: + return packet(CANCEL if cancel else END, {"binding": binding(start)}) + + +def gap_packet(start: StreamStart, *, modality: str, reason: str, count: int) -> bytes: + if modality not in PAYLOAD_LIMITS or reason not in ("unavailable", "source-gap", "overload"): + raise StreamWireError("invalid gap notice") + return packet( + GAP, + { + "binding": binding(start), + "modality": modality, + "reason": reason, + "count": decimal(count), + }, + ) diff --git a/tests/test_perception_streaming_ingress.py b/tests/test_perception_streaming_ingress.py new file mode 100644 index 0000000..5ca5268 --- /dev/null +++ b/tests/test_perception_streaming_ingress.py @@ -0,0 +1,313 @@ +"""Bounded synthetic IPC fixtures, no model or local load tests.""" + +import json +import socket +import threading +from contextlib import contextmanager +from dataclasses import replace + +import pytest + +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_ingress import StreamingIngress +from k1link.perception.streaming_lifecycle import StreamingLifecycle +from k1link.perception.streaming_queue import StreamMailbox +from k1link.perception.streaming_sender import StreamingSender + + +def start(): + return StreamStart( + run_id="test-run", + source_id="recorded-or-live", + worker_id="worker-006", + epoch_id="epoch-1", + lease_generation=1, + profile_sha256="a" * 64, + image_sha256="b" * 64, + effective_config_sha256="c" * 64, + calibration_sha256="d" * 64, + clock_domain_id="original-host-clock", + input_mode="live", + ) + + +def event(seq=1, *, modality="lidar", payload=b"points", source_sequence=None): + return LiveIngressEvent( + ingress_sequence=seq, + session_id="capture-1", + session_generation=1, + modality=modality, + source_id=f"raw/{modality}", + source_sequence=seq if source_sequence is None else source_sequence, + captured_at_epoch_ns=1_799_999_999_123_456_789, + received_monotonic_ns=9_007_199_254_740_993 + seq, + payload=payload, + ) + + +def send_event(sock, observation, *, identity=None): + for header, payload in wire.event_packets(identity or start(), observation): + sock.sendall(header) + sock.sendall(payload) + + +@contextmanager +def harness(tmp_path, *, consume=None, byte_limit=16 * 1024 * 1024, clock=None, **kwargs): + mailbox = StreamMailbox(byte_limit=byte_limit) + run = StreamingLifecycle( + start(), + tmp_path, + mailbox, + threading.Event(), + **({"clock_ns": clock} if clock else {}), + ) + left, right = socket.socketpair() + left.settimeout(1) + received, notices = [], [] + run.ready() + receiver = StreamingIngress( + right, run, "capture-1", 1, consume or received.append, notices.append, **kwargs + ) + receiver.start() + try: + yield left, receiver, run, received, notices + finally: + left.close() + run.request_stop("cancelled") + assert receiver.join() + assert run.close() + + +def hello(sock): + sock.sendall(wire.open_packet(start(), "capture-1", 1)) + + +def test_unknown_duration_delivers_before_end_and_preserves_every_raw_byte(tmp_path): + delivered = threading.Event() + seen = [] + + def consume(value): + seen.append(value) + delivered.set() + + with harness(tmp_path, consume=consume) as (sock, receiver, run, _, _): + hello(sock) + original = event(payload=b"x" * (wire.MAX_FRAGMENT + 19)) + send_event(sock, original) + assert delivered.wait(1) + assert receiver.thread.is_alive() and not run.mailbox.done + assert seen == [original] # Includes int64 values above JS exact-number range. + sock.sendall(wire.terminal_packet(start())) + assert receiver.join() + assert receiver.terminal == "end" and receiver.error is None + assert receiver.counts["fragments"] == 2 + assert receiver.snapshot()["incomplete_observations"] == 0 + assert run.mailbox.bytes == 0 + assert ( + run.mailbox.peak_bytes + <= 2 * len(original.payload) + 3 * wire.MAX_HEADER + wire.PREFIX.size + ) + + +@pytest.mark.parametrize("value", [0, 1, (1 << 53) + 1, (1 << 64) - 1]) +def test_uint64_roundtrip(value): + assert wire.uint64(wire.decimal(value)) == value + + +@pytest.mark.parametrize("value", [1, True, "-1", "01", "1.0", str(1 << 64), "1e6"]) +def test_uint64_rejects_lossy_or_ambiguous_projection(value): + with pytest.raises(wire.StreamWireError): + wire.uint64(value) + + +def test_duplicate_json_fields_and_unbounded_headers_fail_before_body(tmp_path): + with pytest.raises(wire.StreamWireError): + wire.parse_header(bytearray(b'{"binding":"a","binding":"b"}')) + with harness(tmp_path) as (sock, receiver, run, _, _): + sock.sendall(wire.PREFIX.pack(wire.MAGIC, wire.OPEN, wire.MAX_HEADER + 1)) + assert receiver.join() + assert "header size" in receiver.error + assert run.state == GraphState.RUNNING # Unbound peer cannot stop owner. + assert run.mailbox.bytes == 0 + + +@pytest.mark.parametrize( + "field,value", [("epoch_id", "old"), ("lease_generation", 2), ("image_sha256", "f" * 64)] +) +def test_wrong_handshake_does_not_cancel_current_owner(tmp_path, field, value): + with harness(tmp_path) as (sock, receiver, run, _, _): + sock.sendall(wire.open_packet(replace(start(), **{field: value}), "capture-1", 1)) + assert receiver.join() + assert not receiver.opened and run.state == GraphState.RUNNING + run.check_current(start()) + + +@pytest.mark.parametrize( + "mutation,expected", + [ + (lambda h: h.update(binding="e" * 64), "another StreamStart"), + (lambda h: h.update(total_bytes=2 * wire.MAX_FRAGMENT + 1), "length"), + (lambda h: h.update(fragment_bytes=True), "length"), + (lambda h: h.update(offset=1), "length"), + (lambda h: h.update(payload_sha256="e" * 64), "observation integrity"), + (lambda h: h.update(fragment_sha256="e" * 64), "fragment integrity"), + (lambda h: h["event"].update(session_generation="2"), "acquisition"), + (lambda h: h["event"].update(received_monotonic_ns=9007199254740993), "decimal string"), + ], +) +def test_invalid_fragments_fail_closed_without_delivery(tmp_path, mutation, expected): + with harness(tmp_path) as (sock, receiver, run, received, _): + hello(sock) + header, payload = next(wire.event_packets(start(), event())) + value = json.loads(header[wire.PREFIX.size :]) + mutation(value) + sock.sendall(wire.packet(wire.FRAGMENT, value) + payload) + assert receiver.join() + assert expected in receiver.error and not received + assert run.state == GraphState.STOPPING and run.mailbox.bytes == 0 + + +@pytest.mark.parametrize("case", ["truncated", "interleaved", "end", "timeout"]) +def test_incomplete_reassembly_has_terminal_accounting_and_releases_bytes(tmp_path, case): + with harness(tmp_path, fragment_timeout=0.05) as (sock, receiver, run, received, _): + hello(sock) + pieces = list(wire.event_packets(start(), event(payload=b"x" * (wire.MAX_FRAGMENT + 1)))) + sock.sendall(pieces[0][0]) + sock.sendall(pieces[0][1]) + if case == "truncated": + sock.shutdown(socket.SHUT_WR) + elif case == "interleaved": + send_event(sock, event(seq=2)) + elif case == "end": + sock.sendall(wire.terminal_packet(start())) + assert receiver.join() + assert receiver.terminal == "failed" and not received + assert receiver.snapshot()["incomplete_observations"] == 1 + assert run.mailbox.bytes == 0 and run.mailbox.quiescent + + +def test_raw_and_decoded_work_share_one_budget(tmp_path): + # Small bound proves rejection without creating large local pressure. + with harness(tmp_path, byte_limit=3 * wire.MAX_HEADER + 100) as ( + sock, + receiver, + run, + received, + _, + ): + active = {"sequence": 1, "payload_bytes": 85} + assert run.admit(start(), active) + assert run.mailbox.take() is active + hello(sock) + send_event(sock, event()) # Needs 12 extra bytes, only 6 remain. + assert receiver.join() + assert "byte budget" in receiver.error and not received + assert run.mailbox.bytes == 85 and not run.mailbox.quiescent + run.mailbox.release(active) + + +def test_stop_cannot_release_borrowed_payload_until_callback_exits(tmp_path): + entered, leave = threading.Event(), threading.Event() + + def consume(_value): + entered.set() + assert leave.wait(2) + + with harness(tmp_path, consume=consume) as (sock, receiver, run, _, _): + hello(sock) + send_event(sock, event()) + assert entered.wait(1) + try: + assert not run.close("cancelled") + assert run.mailbox.bytes > 0 and not run.lease.released + finally: + leave.set() + assert receiver.join() and run.mailbox.bytes == 0 + assert run.close() and run.lease.released + + +def test_lease_expiry_wakes_idle_receiver_without_new_bytes(tmp_path): + now = [100] + with harness(tmp_path, clock=lambda: now[0]) as (sock, receiver, run, _, _): + hello(sock) + now[0] += 3_000_000_000 + assert receiver.join() + assert run.stop_event.is_set() and run.reason == "lease-lost" + assert run.mailbox.bytes == 0 + + +def test_gaps_reset_camera_codec_and_cancel_is_not_successful_end(tmp_path): + with harness(tmp_path) as (sock, receiver, run, received, notices): + hello(sock) + send_event(sock, event(1, modality="camera-init")) + send_event(sock, event(2, modality="camera-frame")) + sock.sendall( + wire.gap_packet(start(), modality="camera-frame", reason="source-gap", count=1) + ) + send_event(sock, event(4, modality="camera-init")) + send_event(sock, event(5, modality="camera-frame")) + sock.sendall(wire.terminal_packet(start(), cancel=True)) + assert receiver.join() + assert receiver.terminal == "cancelled" and len(received) == 4 + assert len(notices) == 1 and receiver.counts["declared_gap_observations"] == 1 + assert receiver.counts["ingress_sequence_gaps"] == 1 + assert run.reason == "cancelled" + + +@pytest.mark.parametrize("case", ["no-init", "duplicate", "camera-gap", "clock-regression"]) +def test_source_order_and_codec_prerequisites_are_enforced(tmp_path, case): + with harness(tmp_path) as (sock, receiver, _, received, _): + hello(sock) + if case != "no-init": + send_event(sock, event(1, modality="camera-init")) + send_event(sock, event(2, modality="camera-frame")) + bad = event( + 2 if case == "duplicate" else 3, + modality="camera-frame", + source_sequence=4 if case == "camera-gap" else None, + ) + if case == "clock-regression": + bad = replace(bad, received_monotonic_ns=1) + send_event(sock, bad) + assert receiver.join() and receiver.terminal == "failed" + assert len(received) == (0 if case == "no-init" else 2) + + +def test_external_reservation_is_not_freed_by_cancel_and_cannot_double_release(): + queue = StreamMailbox(byte_limit=100) + raw = queue.reserve_ingress(20) + queue.cancel() + assert not queue.quiescent and queue.bytes == 20 + raw.release() + assert queue.quiescent and queue.bytes == 0 + with pytest.raises(ValueError, match="ownership"): + raw.release() + + +def test_sender_streams_existing_events_and_explicit_gap_then_end(tmp_path): + with harness(tmp_path) as (sock, receiver, run, received, notices): + sender = StreamingSender(sock, start(), "capture-1", 1, lambda: run.check_current(start())) + sender.send(event()) + sender.gap(modality="pose", reason="unavailable", count=0) + sender.send(event(2, modality="pose")) + sender.end() + assert receiver.join() and receiver.terminal == "end" + assert len(received) == 2 and notices[0]["reason"] == "unavailable" + with pytest.raises(wire.StreamWireError, match="closed"): + sender.send(event(3)) + + +def test_sender_timeout_closes_transport_instead_of_building_backlog(): + left, right = socket.socketpair() + left.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096) + try: + sender = StreamingSender(left, start(), "capture-1", 1, lambda: None, timeout=0.01) + with pytest.raises(wire.StreamWireError, match="deadline"): + sender.send(event(payload=b"x" * 65536)) + assert sender.closed and left.fileno() == -1 + finally: + left.close() + right.close()