From c31c46caa5fa21c5c6006461e8367e6f6d6e1de5 Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 15:27:06 +0300 Subject: [PATCH] feat(perception): preserve resident runtime across recoverable input gaps --- .../pilot_continuity_probe.py | 281 ++++++++++++++ .../pilot_fragment_decoder.py | 3 + src/k1link/perception/streaming_continuity.py | 92 +++++ src/k1link/perception/streaming_decoder.py | 12 + .../perception/streaming_decoder_client.py | 14 + src/k1link/perception/streaming_ingress.py | 62 ++- src/k1link/perception/streaming_lifecycle.py | 163 +++++++- src/k1link/perception/streaming_queue.py | 20 +- src/k1link/perception/streaming_scheduler.py | 9 +- src/k1link/perception/streaming_sensors.py | 9 + src/k1link/perception/worker_readiness.py | 33 +- tests/test_perception_binary_bridge.py | 7 +- tests/test_perception_streaming_continuity.py | 353 ++++++++++++++++++ tests/test_perception_streaming_decoder.py | 16 + 14 files changed, 1041 insertions(+), 33 deletions(-) create mode 100644 experiments/perception/worker/streaming_profile_stage1/pilot_continuity_probe.py create mode 100644 src/k1link/perception/streaming_continuity.py create mode 100644 tests/test_perception_streaming_continuity.py diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_continuity_probe.py b/experiments/perception/worker/streaming_profile_stage1/pilot_continuity_probe.py new file mode 100644 index 0000000..7dc49d2 --- /dev/null +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_continuity_probe.py @@ -0,0 +1,281 @@ +"""CPU-only recovery proof: real H264 fixture, synthetic clock/pose/points. + +Not a full graph run: a resident CPU sentinel stands in for loaded GPU models. +The real decoder process must survive two IPC reconnects and reinitialize its +codec. The source clock/tick counter continues through each outage, no backlog. +""" + +import argparse +import hashlib +import json +import os +import select +import socket +import struct +import subprocess +import sys +import tempfile +import threading +import time +from datetime import UTC, datetime +from pathlib import Path + +from k1link.compute.live_perception import LiveIngressEvent +from k1link.perception.realtime_contract import StreamStart +from k1link.perception.streaming_continuity import ResumeEvidence, StreamSuspended +from k1link.perception.streaming_decoder_client import StreamingDecoderClient +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 +from k1link.perception.streaming_sensors import CausalSensorWindow, normalized_sensor +from k1link.perception.worker_lease import WorkerLeaseError + + +def run(output, camera): + output.mkdir() + with (camera / "index.jsonl").open() as stream: + raw = stream.readline(65537) + assert len(raw) <= 65536 + row = json.loads(raw) + frame_path = (camera / row["path"]).resolve(strict=True) + assert frame_path.is_relative_to(camera.resolve()) + assert 0 < frame_path.stat().st_size <= 1024 * 1024 + assert 0 < (camera / "init.mp4").stat().st_size <= 65536 + fragment, init = frame_path.read_bytes(), (camera / "init.mp4").read_bytes() + assert len(init) <= 65536 and hashlib.sha256(fragment).hexdigest() == row["sha256"] + report = { + "created_utc": datetime.now(UTC).isoformat(), + "started_monotonic_ns": time.monotonic_ns(), + "scope": ( + "real decoder and Linux processes; synthetic source timestamps/pose/points; " + "NO GPU models or full graph" + ), + "source_full_prepass": False, + "source_index_rows_read": 1, + "fixture_fragment_sha256": hashlib.sha256(fragment).hexdigest(), + "gpu_devices": sorted(str(p) for p in Path("/dev").glob("nvidia*")), + "cycles": [], + } + assert not report["gpu_devices"] and os.environ.get("CUDA_VISIBLE_DEVICES") == "" + activation = StreamStart( + output.name, + "synthetic-recovery", + "synthetic-worker", + "activation-1", + 1, + "a" * 64, + "b" * 64, + hashlib.sha256(b"resumable-input-real-decoder-probe-v1").hexdigest(), + "d" * 64, + "worker-mapped-fixture-clock", + "live", + ) + with tempfile.TemporaryDirectory(prefix="continuity-lease-") as directory: + runtime = StreamingLifecycle( + activation, + Path(directory), + StreamMailbox(), + threading.Event(), + recover_input=True, + source_clock_ns=time.monotonic_ns, + ) + finish = threading.Event() + ticks = [0] + + def controller(): + while not finish.wait(0.05): + ticks[0] += 1 + try: + runtime.renew(activation) + except WorkerLeaseError: + return + + thread = threading.Thread(target=controller, daemon=True) + thread.start() + decoder = window = receiver = sender = None + try: + sentinel = runtime.spawn( + lambda: subprocess.Popen( + [ + sys.executable, + "-c", + "import time; print('ready',flush=True); time.sleep(30)", + ], + stdout=subprocess.PIPE, + start_new_session=True, + ) + ) + assert ( + select.select([sentinel.stdout], [], [], 2)[0] + and sentinel.stdout.readline() == b"ready\n" + ) + child = runtime.spawn( + lambda: subprocess.Popen( + ["python3", "-B", "/probe/pilot_fragment_decoder.py"], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + start_new_session=True, + ) + ) + decoder = StreamingDecoderClient(runtime, child.stdout.fileno(), child.stdin.fileno()) + window = CausalSensorWindow(runtime.mailbox) + runtime.ready() + original_pids = [sentinel.pid, child.pid] + report["resident_pids"] = original_pids + epoch = activation + for cycle in range(3): + received = threading.Event() + item = {"cycle": cycle, "input_start": epoch.to_dict()} + + def consume(event, cycle=cycle, epoch=epoch, item=item, received=received): + if event.modality == "camera-init": + if cycle: + decoder.reset(epoch, event.payload) + window.reset() + else: + decoder.configure(event.payload) + return + if event.modality in ("lidar", "pose"): + window.append( + normalized_sensor( + event.modality, + event.source_sequence, + event.received_monotonic_ns, + event.payload, + ) + ) + return + ticket = runtime.mailbox.reserve_ingress(1440000) + target = None + try: + target = bytearray(1440000) + decoded = decoder.decode(event.payload, target) + binding = window.bind(event.received_monotonic_ns) + assert binding.available and decoded["frame_index"] == 0 + if cycle: + evidence = ResumeEvidence( + event.received_monotonic_ns, + window.pose.time_ns, + binding.increments[0].time_ns, + binding.increments[-1].time_ns, + decoder.frames == 1, + ) + # Synthetic temporal-state owner only; not JointGraph/TGS reset proof. + runtime.resume_input( + epoch, evidence, lambda: item.update(temporal_fixture_reset=True) + ) + with runtime.work(epoch, "gpu"): + item["bgr_sha256"] = hashlib.sha256(target).hexdigest() + runtime.validate_result_binding(epoch.to_dict()) + window.finish_camera(event.received_monotonic_ns) + item["completed_monotonic_ns"] = time.monotonic_ns() + received.set() + finally: + target = None + ticket.release() + + left, right = socket.socketpair() + receiver = StreamingIngress( + right, + runtime, + "fixture-acquisition", + 1, + consume, + lambda _: None, + input_epoch=epoch, + ) + receiver.start() + sender = StreamingSender( + left, + epoch, + "fixture-acquisition", + 1, + lambda epoch=epoch: runtime.check_input(epoch, synchronizing=True), + ) + stamp = time.monotonic_ns() + payloads = [ + ("camera-init", "camera", init), + ("pose", "pose", struct.pack("<7d", 0, 0, 0, 0, 0, 0, 1)), + ("lidar", "points", struct.pack(" 0 + assert sentinel.poll() is None and child.poll() is None + with_rejected = False + try: + runtime.validate_result_binding(epoch.to_dict()) + except StreamSuspended: + with_rejected = True + assert with_rejected + epoch = runtime.begin_input(activation) + item["receiver"] = receiver.snapshot() + item["resident_pids_unchanged"] = original_pids == [sentinel.pid, child.pid] + report["cycles"].append(item) + assert len({item["bgr_sha256"] for item in report["cycles"]}) == 1 + report["passed"] = True + finally: + finish.set() + thread.join(timeout=1) + if sender: + sender.close() + runtime.request_stop("completed" if report.get("passed") else "failed") + if receiver: + receiver.join(timeout=2) + runtime.stop_children() + if window: + window.close() + if decoder: + decoder.close() + report["released"] = runtime.close() + report["runtime"] = runtime.snapshot() + report["input_bytes"] = runtime.mailbox.bytes + report["peak_input_bytes"] = runtime.mailbox.peak_bytes + report["finished_monotonic_ns"] = time.monotonic_ns() + report["source_ticks_total"] = ticks[0] + report["full_graph_recovery_proved"] = False + (output / "report.json").write_text(json.dumps(report, indent=2) + "\n") + for process in runtime._children: + for stream in (process.stdin, process.stdout): + if stream: + stream.close() + assert report["released"] and not report["input_bytes"] + print( + json.dumps({"passed": True, "cycles": 3, "same_decoder_pid": True, "full_graph": False}) + ) + + +if __name__ == "__main__": + parser = argparse.ArgumentParser() + parser.add_argument("--output", required=True) + parser.add_argument("--camera", required=True) + args = parser.parse_args() + run(Path(args.output), Path(args.camera)) diff --git a/experiments/perception/worker/streaming_profile_stage1/pilot_fragment_decoder.py b/experiments/perception/worker/streaming_profile_stage1/pilot_fragment_decoder.py index 01ec72b..765aebd 100644 --- a/experiments/perception/worker/streaming_profile_stage1/pilot_fragment_decoder.py +++ b/experiments/perception/worker/streaming_profile_stage1/pilot_fragment_decoder.py @@ -20,6 +20,9 @@ def main(): if header == {"op": "init"}: decoder.configure(raw) send(sys.stdout.buffer, {"initialized": True}) + elif header == {"op": "reset"}: + decoder.reset(raw) + send(sys.stdout.buffer, {"reset": True}) elif header == {"op": "decode"}: image = decoder.decode(raw) send( diff --git a/src/k1link/perception/streaming_continuity.py b/src/k1link/perception/streaming_continuity.py new file mode 100644 index 0000000..94eb413 --- /dev/null +++ b/src/k1link/perception/streaming_continuity.py @@ -0,0 +1,92 @@ +"""Recoverable input epochs; resident model ownership is a separate lifetime. + +Used only by the explicitly enabled resumable adapter. The controller supplies +the source-clock mapping and the profile adapter supplies verified decoder and +sensor evidence. No socket, GPU, vehicle command or unbounded history lives here. +""" + +from __future__ import annotations + +from dataclasses import dataclass, replace +from hashlib import sha256 + +from .realtime_contract import StreamStart, _integer + + +class StreamSuspended(RuntimeError): + """Discard obsolete work, keep the resident model processes and lease.""" + + +@dataclass(frozen=True, slots=True) +class ResumeEvidence: + camera_ns: int + pose_ns: int + oldest_points_ns: int + newest_points_ns: int + decoded_keyframe: bool + + def validate(self, *, cutoff_ns: int, source_now_ns: int) -> None: + for name in ("camera_ns", "pose_ns", "oldest_points_ns", "newest_points_ns"): + _integer(getattr(self, name), name) + _integer(source_now_ns, "source clock") + if ( + self.decoded_keyframe is not True + or not cutoff_ns <= self.pose_ns <= self.camera_ns <= source_now_ns + or not cutoff_ns <= self.oldest_points_ns <= self.newest_points_ns <= self.camera_ns + or source_now_ns - self.camera_ns > 250_000_000 + or self.camera_ns - self.pose_ns > 100_000_000 + or self.camera_ns - self.newest_points_ns > 100_000_000 + or self.camera_ns - self.oldest_points_ns > 250_000_000 + or max( + abs(self.pose_ns - self.oldest_points_ns), abs(self.pose_ns - self.newest_points_ns) + ) + > 100_000_000 + ): + raise StreamSuspended("fresh keyframe and causal camera/pose/points required") + + +class InputContinuity: + """Access is serialized by StreamingLifecycle's lock.""" + + def __init__(self, activation: StreamStart) -> None: + self.activation = self.epoch = activation + self.phase = "active" + self.generation = self.pauses = 0 + self.reason: str | None = None + self.cutoff_ns = 0 + + def check(self, epoch: StreamStart, *, synchronizing: bool = False) -> None: + if epoch != self.epoch: + raise StreamSuspended("obsolete input epoch") + if self.phase != "active" and not (synchronizing and self.phase == "synchronizing"): + raise StreamSuspended("input is waiting for resynchronization") + + def pause(self, reason: str) -> None: + if reason not in ("input-disconnected", "input-timeout", "source-gap", "worker-telemetry"): + raise ValueError("unknown recoverable pause reason") + if self.phase != "waiting": + self.pauses += 1 + self.phase, self.reason = "waiting", reason + + def begin(self, source_now_ns: int) -> StreamStart: + _integer(source_now_ns, "source clock") + if self.phase != "waiting" or source_now_ns < self.cutoff_ns: + raise StreamSuspended("new input needs a pause and nonregressing source clock") + self.generation += 1 + identity = sha256(f"{self.activation.epoch_id}:{self.generation}".encode()).hexdigest() + # Wire v2 already binds every packet to the complete StreamStart. Change + # the input epoch, NOT the resident lease generation/model activation. + self.epoch = replace(self.activation, epoch_id="input-" + identity) + self.cutoff_ns, self.phase = source_now_ns, "synchronizing" + return self.epoch + + def snapshot(self) -> dict[str, object]: + return { + "phase": self.phase, + "reason": self.reason, + "input_generation": self.generation, + "pauses": self.pauses, + "input_start": self.epoch.to_dict(), + "minimum_source_ns": str(self.cutoff_ns), + "actuation_allowed": False, + } diff --git a/src/k1link/perception/streaming_decoder.py b/src/k1link/perception/streaming_decoder.py index bf8bbe4..488f8ca 100644 --- a/src/k1link/perception/streaming_decoder.py +++ b/src/k1link/perception/streaming_decoder.py @@ -105,6 +105,18 @@ class FragmentDecoder: self.failed = True raise + def reset(self, payload: bytes) -> None: + """New input epoch only: release codec references, retain the child. + + A genuine decoder failure is not revived by reset. The first new frame + still has to independently pass random-access/keyframe validation. + """ + if self.failed: + raise StreamingDecodeError("failed decoder cannot be resumed") + self._codec = self._init = self._timing = None + self._next_dts, self.frames = None, 0 + self.configure(payload) + def close(self) -> None: self.failed = True self._codec = self._init = self._timing = None diff --git a/src/k1link/perception/streaming_decoder_client.py b/src/k1link/perception/streaming_decoder_client.py index 054e788..838be33 100644 --- a/src/k1link/perception/streaming_decoder_client.py +++ b/src/k1link/perception/streaming_decoder_client.py @@ -6,6 +6,7 @@ import math from typing import Any from .graph_contracts import GraphState +from .realtime_contract import StreamStart from .streaming_lifecycle import StreamingLifecycle from .streaming_pipe_rpc import HEADER_SCRATCH, MAX_OUTPUT, BoundedPipeRpc, PipeRpcError @@ -57,6 +58,19 @@ class StreamingDecoderClient: self.frames += 1 return result + def reset(self, epoch: StreamStart, raw: bytes) -> None: + self.runtime.check_input(epoch, synchronizing=True) + if ( + self.closed + or self.runtime.continuity is None + or self.runtime.continuity.phase != "synchronizing" + or not 0 < len(raw) <= 65536 + ): + raise PipeRpcError("decoder reset requires a new input epoch") + if self.rpc.exchange({"op": "reset"}, raw, bytearray()) != {"reset": True}: + raise PipeRpcError("decoder reset response changed") + self.frames = 0 + def close(self) -> None: # Caller joins its callback and stops the owned child before closing. if not self.closed: diff --git a/src/k1link/perception/streaming_ingress.py b/src/k1link/perception/streaming_ingress.py index feefce8..c95372d 100644 --- a/src/k1link/perception/streaming_ingress.py +++ b/src/k1link/perception/streaming_ingress.py @@ -1,7 +1,9 @@ """One controller-supplied IPC socket feeding the common streaming lifecycle. -No listener, acquisition, archive reader, model RPC, network authentication or -automatic reconnect. Reconnect needs a new StreamStart/lease and decoder state. +No listener, acquisition, archive reader, model RPC or network authentication. +An explicitly resumable runtime preserves its resident lease on disconnect; +the controller supplies a new input epoch/socket and the adapter resets decoder +and temporal state. The historical non-resumable diagnostic remains supported. The synchronous consumer is a trusted adapter: raw events are borrowed for that call only. It must reserve decoder scratch before allocation and retain decoded work only through the common mailbox. Slow/failed consumers fail this stream; @@ -16,15 +18,22 @@ import threading import time from collections import Counter from collections.abc import Callable +from contextlib import suppress from hashlib import sha256 from typing import Any from k1link.compute.live_perception import LiveIngressEvent from . import streaming_wire as wire +from .realtime_contract import StreamStart +from .streaming_continuity import StreamSuspended from .streaming_lifecycle import StreamingLifecycle +class InputInterrupted(wire.StreamWireError): + """Transport unavailable, not malformed model data or an ownership loss.""" + + class StreamingIngress: def __init__( self, @@ -37,6 +46,7 @@ class StreamingIngress: *, fragment_timeout: float = 0.25, idle_timeout: float = 2.0, + input_epoch: StreamStart | None = None, ) -> None: if not 0 < fragment_timeout <= 2 or not 0 < idle_timeout <= 30: raise ValueError("invalid bounded ingress timeouts") @@ -49,7 +59,8 @@ class StreamingIngress: 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.input_epoch = input_epoch or runtime.start + self.binding = wire.binding(self.input_epoch) self.opened = False self.error: str | None = None self.terminal: str | None = None @@ -60,10 +71,10 @@ class StreamingIngress: self.thread = threading.Thread( target=self._serve, name="perception-binary-ingress", daemon=True ) - runtime.track_thread(self.thread) + runtime.register_ingress(self.thread, self.input_epoch) def start(self) -> None: - self.runtime.check_current(self.runtime.start) + self.runtime.check_input(self.input_epoch, synchronizing=True) self.thread.start() def join(self, timeout: float = 1.0) -> bool: @@ -71,9 +82,9 @@ class StreamingIngress: return not self.thread.is_alive() def _check(self, deadline: float) -> None: - self.runtime.check_current(self.runtime.start) + self.runtime.check_input(self.input_epoch, synchronizing=True) if time.monotonic() >= deadline: - raise wire.StreamWireError("ingress deadline exceeded") + raise InputInterrupted("ingress deadline exceeded") def _read_into(self, target: memoryview, deadline: float) -> None: offset = 0 @@ -87,8 +98,10 @@ class StreamingIngress: read = self.connection.recv_into(target[offset:]) except BlockingIOError: continue + except (ConnectionResetError, ConnectionAbortedError, BrokenPipeError) as exc: + raise InputInterrupted("input connection lost") from exc if read == 0: - raise wire.StreamWireError("unexpected EOF without explicit End") + raise InputInterrupted("unexpected EOF without explicit End") self.counts["wire_bytes"] += read offset += read self._check(deadline) @@ -109,8 +122,10 @@ class StreamingIngress: def _bound(self, header: dict[str, Any]) -> None: if header.get("binding") != self.binding: + if self.runtime.continuity is not None: + raise StreamSuspended("obsolete packet from another input epoch") raise wire.StreamWireError("frame belongs to another StreamStart") - self.runtime.check_current(self.runtime.start) + self.runtime.check_input(self.input_epoch, synchronizing=True) def _fragment(self, header: dict[str, Any]) -> tuple[dict[str, Any], int, int, int]: if set(header) != { @@ -157,6 +172,8 @@ class StreamingIngress: 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: + if self.runtime.continuity is not None: + raise StreamSuspended("camera sequence gap requires resynchronization") raise wire.StreamWireError("camera sequence gap requires new codec initialization") self.counts["observations_started"] += 1 reservation = self.runtime.mailbox.reserve_ingress(2 * total) @@ -191,6 +208,10 @@ class StreamingIngress: if sha256(raw).hexdigest() != first["payload_sha256"]: raise wire.StreamWireError("observation integrity mismatch") self._check(deadline) + if event["modality"] != "camera-init": + self.runtime.check_input( + self.input_epoch, synchronizing=True, event_ns=event["received_monotonic_ns"] + ) admitted = LiveIngressEvent(**event, payload=bytes(raw)) raw = None # Immutable callback bytes remain fully reserved. self.consume(admitted) @@ -224,8 +245,10 @@ class StreamingIngress: self.counts["declared_gap_observations"] += wire.uint64(header["count"]) if header["modality"].startswith("camera"): self.camera_initialized = False + if self.runtime.continuity is not None: + raise StreamSuspended("declared source gap requires resynchronization") self.notice(dict(header)) - self.runtime.check_current(self.runtime.start) + self.runtime.check_input(self.input_epoch, synchronizing=True) self.counts["gap_notices"] += 1 def _serve(self) -> None: @@ -240,7 +263,7 @@ class StreamingIngress: expected = wire.parse_header( bytearray( wire.open_packet( - self.runtime.start, + self.input_epoch, self.session_id, self.session_generation, )[wire.PREFIX.size :] @@ -269,8 +292,21 @@ class StreamingIngress: self.error = str(exc) self.terminal = "failed" if self.opened: - self.runtime.mailbox.finish(self.error) - self.runtime.request_stop("failed") + if self.runtime.continuity is not None and isinstance( + exc, (InputInterrupted, StreamSuspended) + ): + self.terminal = "paused" + # A late old socket cannot pause a replacement connection. + with suppress(RuntimeError): # Already replaced, stopping, or fenced. + self.runtime.pause_input( + self.input_epoch, + "input-disconnected" + if isinstance(exc, InputInterrupted) + else "source-gap", + ) + else: + self.runtime.mailbox.finish(self.error) + self.runtime.request_stop("failed") finally: self.connection.close() if reservation is not None: diff --git a/src/k1link/perception/streaming_lifecycle.py b/src/k1link/perception/streaming_lifecycle.py index 163b704..c073314 100644 --- a/src/k1link/perception/streaming_lifecycle.py +++ b/src/k1link/perception/streaming_lifecycle.py @@ -21,10 +21,15 @@ from typing import Any, Literal from .graph_contracts import GraphState from .realtime_contract import StreamStart from .realtime_scene import _wire_integer +from .streaming_continuity import InputContinuity, ResumeEvidence, StreamSuspended from .streaming_queue import IngressReservation, StreamBundle, StreamMailbox from .worker_lease import WorkerLease, WorkerLeaseError from .worker_operating_envelope import WorkerSnapshot -from .worker_readiness import WorkerReadinessError, WorkerReadinessMonitor +from .worker_readiness import ( + WorkerReadinessError, + WorkerReadinessMonitor, + WorkerTelemetryUnavailable, +) def _group_exists(pgid: int) -> bool: @@ -48,16 +53,25 @@ class StreamingLifecycle: ttl_seconds: float = 2.0, clock_ns: Callable[[], int] = time.monotonic_ns, readiness: WorkerReadinessMonitor | None = None, + recover_input: bool = False, + source_clock_ns: Callable[[], int] | None = None, ) -> None: self.start, self.mailbox, self.stop_event = start, mailbox, stop self._clock_ns, self._readiness = clock_ns, readiness + if recover_input and ( + source_clock_ns is None or (readiness is not None and not readiness.recoverable) + ): + raise ValueError("resumable input needs mapped source clock and recoverable telemetry") + self.continuity = InputContinuity(start) if recover_input else None + self._source_clock_ns = source_clock_ns + self._ingress_thread: threading.Thread | None = None if readiness is not None: readiness.check(start, now_monotonic_ns=clock_ns(), require_warmup=False) self._lock = threading.RLock() self._cleanup_lock = threading.Lock() self._children: list[subprocess.Popen[bytes]] = [] self._threads: list[threading.Thread] = [] - self._active = {"gpu": 0, "cpu": 0} + self._active = {"gpu": 0, "cpu": 0, "resync": 0} self.state = GraphState.CREATED self.reason: str | None = None self.stop_requested_ns: int | None = None @@ -70,7 +84,9 @@ class StreamingLifecycle: ) self._watchdog.start() - def _check(self, start: StreamStart, *, starting: bool = False) -> None: + def _check( + self, start: StreamStart, *, starting: bool = False, allow_unavailable: bool = False + ) -> None: allowed = (GraphState.STARTING, GraphState.RUNNING) if starting else (GraphState.RUNNING,) if self.state not in allowed or self.stop_event.is_set(): raise WorkerLeaseError("runtime is not accepting work") @@ -81,9 +97,11 @@ class StreamingLifecycle: if start == self.start: self.request_stop("lease-lost") raise - self._check_readiness(require_warmup=self.state == GraphState.RUNNING) + self._check_readiness( + require_warmup=self.state == GraphState.RUNNING, allow_unavailable=allow_unavailable + ) - def _check_readiness(self, *, require_warmup: bool) -> None: + def _check_readiness(self, *, require_warmup: bool, allow_unavailable: bool = False) -> None: if self._readiness is not None: try: self._readiness.check( @@ -91,6 +109,14 @@ class StreamingLifecycle: now_monotonic_ns=self._clock_ns(), require_warmup=require_warmup, ) + except WorkerTelemetryUnavailable as exc: + if self.continuity is None: + self.request_stop("worker-not-ready") + raise + self.continuity.pause("worker-telemetry") + self.mailbox.pause() + if not allow_unavailable: + raise StreamSuspended(str(exc)) from exc except WorkerReadinessError: self.request_stop("worker-not-ready") raise @@ -102,7 +128,7 @@ class StreamingLifecycle: heartbeat are independent requirements. No host operations under lock. """ with self._lock: - self._check(start, starting=True) + self._check(start, starting=True, allow_unavailable=True) if self._readiness is None: raise WorkerReadinessError("worker readiness monitoring is not configured") try: @@ -112,13 +138,19 @@ class StreamingLifecycle: now_monotonic_ns=self._clock_ns(), require_warmup=self.state == GraphState.RUNNING, ) + except WorkerTelemetryUnavailable: + if self.continuity is None: + self.request_stop("worker-not-ready") + raise + self.continuity.pause("worker-telemetry") + self.mailbox.pause() except WorkerReadinessError: self.request_stop("worker-not-ready") raise def renew(self, start: StreamStart) -> None: with self._lock: - self._check(start, starting=True) + self._check(start, starting=True, allow_unavailable=True) self.lease.renew(start) def ready(self) -> None: @@ -148,32 +180,127 @@ class StreamingLifecycle: def track_thread(self, thread: threading.Thread) -> None: with self._lock: self._check(self.start, starting=True) + self._threads = [t for t in self._threads if t.ident is None or t.is_alive()] if len(self._threads) >= 8 or thread in self._threads: raise WorkerLeaseError("runtime thread registration outside bound") self._threads.append(thread) - def admit(self, start: StreamStart, bundle: StreamBundle) -> bool: + def register_ingress(self, thread: threading.Thread, epoch: StreamStart) -> None: + with self._lock: + if self.continuity is not None: + self.check_input(epoch, synchronizing=True) + if self._ingress_thread is not None and ( + self._ingress_thread.ident is None or self._ingress_thread.is_alive() + ): + raise StreamSuspended("one input connection per resident profile") + self.track_thread(thread) + self._ingress_thread = thread + + def check_input( + self, epoch: StreamStart, *, synchronizing: bool = False, event_ns: int | None = None + ) -> None: + with self._lock: + if self.continuity is None: + self._check(epoch) + return + self._check(self.start) + self.continuity.check(epoch, synchronizing=synchronizing) + if event_ns is not None: + assert self._source_clock_ns is not None + now = self._source_clock_ns() + if event_ns < self.continuity.cutoff_ns or not 0 <= now - event_ns <= 250_000_000: + raise StreamSuspended("obsolete or future input, no backlog replay") + + def pause_input(self, epoch: StreamStart, reason: str) -> None: + with self._lock: + if self.continuity is None: + raise ValueError("resumable input was not enabled") + # A disconnected old socket must not pause the replacement epoch. + if epoch != self.continuity.epoch: + raise StreamSuspended("obsolete input epoch") + self._check(self.start, allow_unavailable=True) + self.continuity.pause(reason) + self.mailbox.pause() + + def begin_input(self, start: StreamStart) -> StreamStart: with self._lock: self._check(start) + if self.continuity is None or self._source_clock_ns is None: + raise ValueError("resumable input was not enabled") + if any(self._active.values()) or ( + self._ingress_thread is not None and self._ingress_thread.is_alive() + ): + raise StreamSuspended("old epoch callbacks or connection still active") + if self.continuity.phase != "waiting": + raise StreamSuspended("pause required before a new input epoch") + self.mailbox.begin_epoch() + if self._ingress_thread is not None: + self._threads = [t for t in self._threads if t is not self._ingress_thread] + self._ingress_thread = None + return self.continuity.begin(self._source_clock_ns()) + + def resume_input( + self, epoch: StreamStart, evidence: ResumeEvidence, reset_temporal: Callable[[], None] + ) -> None: + """Trusted adapter proves a new decoded keyframe + current sensor pair. + + Reset only temporal/rolling outputs, outside the lock, after old compute + has drained. No model restart. A failed reset is an actual runtime fault. + """ + with self._lock: + self.check_input(epoch, synchronizing=True) + if ( + self.continuity is None + or self.continuity.phase != "synchronizing" + or any(self._active.values()) + ): + raise StreamSuspended("resynchronization is not quiescent") + assert self._source_clock_ns is not None + evidence.validate( + cutoff_ns=self.continuity.cutoff_ns, source_now_ns=self._source_clock_ns() + ) + self._active["resync"] = 1 + try: + reset_temporal() + with self._lock: + self.check_input(epoch, synchronizing=True) + evidence.validate( + cutoff_ns=self.continuity.cutoff_ns, source_now_ns=self._source_clock_ns() + ) + self.continuity.phase, self.continuity.reason = "active", None + except StreamSuspended: + raise + except BaseException: + self.request_stop("failed") + raise + finally: + with self._lock: + self._active["resync"] = 0 + + def admit(self, start: StreamStart, bundle: StreamBundle) -> bool: + with self._lock: + self.check_input(start) return self.mailbox.put(bundle) def admit_reserved( self, start: StreamStart, bundle: StreamBundle, reservation: IngressReservation ) -> bool: with self._lock: - self._check(start) + self.check_input(start) return self.mailbox.put_reserved(bundle, reservation) @contextmanager def work(self, start: StreamStart, lane: Literal["gpu", "cpu"]) -> Iterator[None]: with self._lock: - self._check(start) + self.check_input(start) if lane not in self._active or self._active[lane]: raise WorkerLeaseError("only one active call per compute lane") self._active[lane] += 1 try: yield - self.check_current(start) + self.check_input(start) + except StreamSuspended: + raise # The caller discards this epoch's result, not its models. except BaseException: self.request_stop("failed") raise @@ -182,14 +309,17 @@ class StreamingLifecycle: self._active[lane] -= 1 def check_current(self, start: StreamStart, *, starting: bool = False) -> None: - """Only child startup handshakes may opt into the warmup state.""" + """Resource/IPC guard: drain active RPC on pause; work/result gates fence it. + + Only child startup handshakes may opt into the warmup state. + """ with self._lock: - self._check(start, starting=starting) + self._check(start, starting=starting, allow_unavailable=self.continuity is not None) def validate_result_binding(self, value: object) -> None: """Recheck at receipt/use; accepting a hash alone cannot renew authority.""" start = StreamStart.from_dict(value) - self.check_current(start) + self.check_input(start) def request_stop(self, reason: str = "cancelled") -> None: if reason not in ( @@ -217,7 +347,7 @@ class StreamingLifecycle: try: with self._lock: if self.state in (GraphState.STARTING, GraphState.RUNNING): - self._check(self.start, starting=True) + self._check(self.start, starting=True, allow_unavailable=True) if any(p.poll() is not None for p in self._children): self.request_stop("child-exited") stopping = self.state == GraphState.STOPPING @@ -304,6 +434,9 @@ class StreamingLifecycle: _wire_integer(self.retired_ns) if self.retired_ns is not None else None ), "event_clock": "worker-process-monotonic", + "input_continuity": self.continuity.snapshot() + if self.continuity is not None + else {"enabled": False}, "worker_readiness": ( self._readiness.snapshot() if self._readiness is not None diff --git a/src/k1link/perception/streaming_queue.py b/src/k1link/perception/streaming_queue.py index 83fd6e2..0e2dd71 100644 --- a/src/k1link/perception/streaming_queue.py +++ b/src/k1link/perception/streaming_queue.py @@ -14,7 +14,7 @@ from threading import Condition from typing import Any StreamBundle = Mapping[str, Any] -DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed")) +DISCARD_REASONS = frozenset(("cancelled", "gpu-failed", "processing-failed", "input-gap")) class IngressReservation: @@ -210,3 +210,21 @@ class StreamMailbox: while self.pending: self._discard_pending(reason) self.condition.notify_all() + + def pause(self) -> None: + """Drop queued work without EOF, freeing only queue-owned payloads.""" + with self.condition: + while self.pending: + self._discard_pending("input-gap") + self.condition.notify_all() + + def begin_epoch(self) -> None: + """Only after all prior queued/active/completed bundles are released. + + Long-lived bounded decoder/sensor scratch reservations may remain. + The lifecycle separately verifies that the old ingress thread is gone. + """ + with self.condition: + if self.done or self._owned or self.external_pending: + raise ValueError("old epoch still owns work or mailbox is closed") + self._last_sequence = -1 diff --git a/src/k1link/perception/streaming_scheduler.py b/src/k1link/perception/streaming_scheduler.py index e996bbe..20e7ceb 100644 --- a/src/k1link/perception/streaming_scheduler.py +++ b/src/k1link/perception/streaming_scheduler.py @@ -13,6 +13,7 @@ import traceback from collections.abc import Callable from typing import Any +from .streaming_continuity import StreamSuspended from .streaming_queue import StreamBundle, StreamMailbox @@ -43,7 +44,13 @@ class SerialGpuStage: break if self.stop.is_set(): break - result = self.compute(active) + try: + result = self.compute(active) + except StreamSuspended: + self.mailbox.release(active, discard_reason="input-gap") + active = None + self.output_slot.release() + continue with self.output_condition: if self.stop.is_set(): break diff --git a/src/k1link/perception/streaming_sensors.py b/src/k1link/perception/streaming_sensors.py index e2320c4..3937410 100644 --- a/src/k1link/perception/streaming_sensors.py +++ b/src/k1link/perception/streaming_sensors.py @@ -246,3 +246,12 @@ class CausalSensorWindow: self.pose = None self.closed = True self._reservation.release() + + def reset(self) -> None: + """Called by the single adapter after old-epoch callbacks have drained.""" + if self.closed: + raise ValueError("closed sensor window cannot resume") + self.rolling.clear() + self.fresh.clear() + self.pose = self.previous_camera_time = None + self._last_time = -1 diff --git a/src/k1link/perception/worker_readiness.py b/src/k1link/perception/worker_readiness.py index f1bff8b..1dc1e64 100644 --- a/src/k1link/perception/worker_readiness.py +++ b/src/k1link/perception/worker_readiness.py @@ -8,6 +8,7 @@ clocks, or certify real-time operation. The lifecycle serializes all access. from __future__ import annotations +from contextlib import suppress from dataclasses import asdict from typing import Literal @@ -34,6 +35,10 @@ class WorkerReadinessError(WorkerLeaseError): """A current controller lost the prerequisites for further execution.""" +class WorkerTelemetryUnavailable(WorkerReadinessError): + """Readiness is unproved, but no confirmed ownership/runtime failure exists.""" + + class WorkerReadinessMonitor: def __init__( self, @@ -44,12 +49,14 @@ class WorkerReadinessMonitor: mode: ReadinessMode, clock_domain_id: str, now_monotonic_ns: int, + recoverable: bool = False, ) -> None: if mode not in ("strict-envelope", "labelled-experiment"): raise RealtimeContractError("unknown Worker readiness mode") _identifier(clock_domain_id, "worker clock domain") self.start, self.envelope = start, envelope self.mode, self.clock_domain_id = mode, clock_domain_id + self.recoverable = recoverable self._observed = initial self._last_check_ns = now_monotonic_ns self._failure: tuple[str, ...] = () @@ -97,7 +104,26 @@ class WorkerReadinessMonitor: if failure not in _PERFORMANCE_FAILURES and (require_warmup or failure != "warmup-not-complete") ) - if fatal: + if self.recoverable: + # Missing metrics are not a proof that another process owns GPU. + # Known identity/conflict/clock failures still fence immediately. + uncertain = {"worker-snapshot-expired", "warmup-not-complete"} + for field in ("image_sha256", "effective_config_sha256"): + if getattr(self._observed, field) is None: + uncertain.add(f"{field}-mismatch-or-unknown") + if self._observed.competing_gpu_clients is None: + uncertain.add("competing-gpu-clients-or-inventory-unknown") + if self._observed.gpu_owner_run_id in ( + None, + start.run_id, + ) and self._observed.lease_generation in (None, start.lease_generation): + uncertain.add("exclusive-worker-lease-unproved") + confirmed = tuple(reason for reason in fatal if reason not in uncertain) + if confirmed: + self._fail(confirmed) + if fatal or (require_warmup and self.mode == "strict-envelope" and failures): + raise WorkerTelemetryUnavailable("worker readiness unproved: " + ",".join(failures)) + elif fatal: self._fail(fatal) if require_warmup and self.mode == "strict-envelope" and failures: self._fail(failures) @@ -112,7 +138,9 @@ class WorkerReadinessMonitor: ) -> None: # Recheck the OLD observation before replacement: late telemetry cannot # resurrect expired authority, even if the watchdog has not run yet. - self.check(start, now_monotonic_ns=now_monotonic_ns, require_warmup=require_warmup) + # Fresh facts may restore readiness, not the separate lease/input epoch. + with suppress(WorkerTelemetryUnavailable): + self.check(start, now_monotonic_ns=now_monotonic_ns, require_warmup=require_warmup) if observed.observed_monotonic_ns <= self._observed.observed_monotonic_ns: self._fail(("worker-snapshot-not-increasing",)) self._observed = observed @@ -126,6 +154,7 @@ class WorkerReadinessMonitor: return { "enabled": True, "mode": self.mode, + "recoverable": self.recoverable, "envelope": asdict(self.envelope), "worker_clock_domain_id": self.clock_domain_id, "last_check_monotonic_ns": str(self._last_check_ns), diff --git a/tests/test_perception_binary_bridge.py b/tests/test_perception_binary_bridge.py index 2a1e582..14cfa4a 100644 --- a/tests/test_perception_binary_bridge.py +++ b/tests/test_perception_binary_bridge.py @@ -229,7 +229,12 @@ def test_bridge_initialization_failure_releases_unstarted_resources(monkeypatch, monkeypatch.setattr(bridge.socket, "socketpair", sockets) # Only constructor ownership is under test, without a live lease/thread. monkeypatch.setattr("k1link.perception.streaming_ingress.wire.binding", lambda start: {}) - runtime = SimpleNamespace(mailbox=mailbox, start=None, track_thread=register) + runtime = SimpleNamespace( + mailbox=mailbox, + start=None, + track_thread=register, + register_ingress=lambda thread, epoch: register(thread), + ) source = SimpleNamespace(source_zero=0, wall_zero=0, run=lambda sock: None) try: with pytest.raises(ValueError, match="injected"): diff --git a/tests/test_perception_streaming_continuity.py b/tests/test_perception_streaming_continuity.py new file mode 100644 index 0000000..10a065e --- /dev/null +++ b/tests/test_perception_streaming_continuity.py @@ -0,0 +1,353 @@ +"""Small synthetic pause/resume tests: never model/GPU load on the Mac.""" + +import socket +import threading +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_continuity import ResumeEvidence, StreamSuspended +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_scheduler import SerialGpuStage +from k1link.perception.worker_lease import WorkerLeaseError +from k1link.perception.worker_operating_envelope import WorkerOperatingEnvelope, WorkerSnapshot +from k1link.perception.worker_readiness import WorkerReadinessMonitor + + +def binding(): + return StreamStart( + "run", + "source", + "worker", + "epoch", + 1, + "a" * 64, + "b" * 64, + "c" * 64, + "d" * 64, + "source-clock", + "live", + ) + + +def snapshot(now, **changes): + return replace( + WorkerSnapshot( + "worker", + "worker-clock", + now, + "GPU", + "driver", + "b" * 64, + "c" * 64, + 8000, + 8192, + 2610, + 10251, + "run", + 1, + (), + True, + ), + **changes, + ) + + +@pytest.fixture +def harness(tmp_path): + runs = [] + + def create(*, telemetry=False): + clock = [1_000_000_000] + monitor = ( + WorkerReadinessMonitor( + binding(), + WorkerOperatingEnvelope("envelope", "GPU", "driver", 8000, 8192, 2610, 10251), + snapshot(clock[0]), + mode="strict-envelope", + clock_domain_id="worker-clock", + now_monotonic_ns=clock[0], + recoverable=True, + ) + if telemetry + else None + ) + run = StreamingLifecycle( + binding(), + tmp_path / "lease", + StreamMailbox(), + threading.Event(), + clock_ns=lambda: clock[0], + recover_input=True, + source_clock_ns=lambda: clock[0], + readiness=monitor, + ) + run.ready() + runs.append(run) + return run, clock + + yield create + for run in runs: + assert run.close() + + +def proof(now): + return ResumeEvidence(now, now, now, now, True) + + +def test_pause_keeps_lease_mailbox_and_warm_models_then_new_epoch(harness): + run, clock = harness() + run.admit(run.start, {"sequence": 100, "payload_bytes": 8}) + run.pause_input(run.start, "input-disconnected") + assert run.state == GraphState.RUNNING and not run.stop_event.is_set() + assert not run.mailbox.done and not run.mailbox.bytes and not run.lease.released + assert run.mailbox.drop_counts["input-gap"] == 1 + for _ in range(4): + clock[0] += 900_000_000 + run.renew(run.start) + with pytest.raises(StreamSuspended): + run.admit(run.start, {"sequence": 101, "payload_bytes": 8}) + epoch = run.begin_input(run.start) + assert epoch.epoch_id != run.start.epoch_id and epoch.lease_generation == 1 + reset = [] + with pytest.raises(StreamSuspended): + run.admit(epoch, {"sequence": 0, "payload_bytes": 8}) + run.resume_input(epoch, proof(clock[0]), lambda: reset.append("temporal-reset")) + assert reset == ["temporal-reset"] and run.continuity.phase == "active" + assert run.admit(epoch, {"sequence": 0, "payload_bytes": 8}) + run.mailbox.release(run.mailbox.take()) + with pytest.raises(StreamSuspended): + run.validate_result_binding(run.start.to_dict()) + run.validate_result_binding(epoch.to_dict()) + + +def test_active_old_callback_cannot_publish_or_be_freed_early(harness): + run, clock = harness() + packet = {"sequence": 0, "payload_bytes": 8} + run.admit(run.start, packet) + assert run.mailbox.take() is packet + with pytest.raises(StreamSuspended), run.work(run.start, "gpu"): + run.pause_input(run.start, "input-timeout") + run.check_current(run.start) # In-flight IPC can drain without poisoning the child. + with pytest.raises(StreamSuspended): + run.begin_input(run.start) + with pytest.raises(ValueError, match="still owns"): + run.begin_input(run.start) + run.mailbox.release(packet, discard_reason="input-gap") + epoch = run.begin_input(run.start) + run.resume_input(epoch, proof(clock[0]), lambda: None) + with pytest.raises(StreamSuspended): + run.pause_input(run.start, "input-timeout") + assert run.continuity.phase == "active" + + +@pytest.mark.parametrize( + "field,value", + [ + ("decoded_keyframe", False), + ("pose_ns", 0), + ("newest_points_ns", 0), + ("oldest_points_ns", 0), + ("camera_ns", 2_000_000_000), + ], +) +def test_resume_requires_fresh_synchronized_evidence(harness, field, value): + run, clock = harness() + run.pause_input(run.start, "source-gap") + epoch = run.begin_input(run.start) + called = [] + with pytest.raises(StreamSuspended): + run.resume_input( + epoch, replace(proof(clock[0]), **{field: value}), lambda: called.append(True) + ) + assert not called and run.continuity.phase == "synchronizing" and not run.stop_event.is_set() + + +def test_pause_during_reset_never_reactivates_input(harness): + run, clock = harness() + run.pause_input(run.start, "source-gap") + epoch = run.begin_input(run.start) + with pytest.raises(StreamSuspended): + run.resume_input( + epoch, proof(clock[0]), lambda: run.pause_input(epoch, "input-disconnected") + ) + assert run.continuity.phase == "waiting" + + +def test_reset_error_is_fatal_not_hidden_as_a_network_pause(harness): + run, clock = harness() + run.pause_input(run.start, "source-gap") + epoch = run.begin_input(run.start) + + def broken(): + raise ValueError("reset failed") + + with pytest.raises(ValueError, match="reset failed"): + run.resume_input(epoch, proof(clock[0]), broken) + assert run.stop_event.is_set() + + +@pytest.mark.parametrize( + "changes", + [ + {"competing_gpu_clients": None}, + {"gpu_owner_run_id": None}, + {"memory_clock_mhz": 405}, + {"image_sha256": None}, + ], +) +def test_telemetry_unknown_or_low_clocks_pause_and_fresh_facts_require_resync(harness, changes): + run, clock = harness(telemetry=True) + clock[0] += 100_000_000 + run.observe_worker(run.start, snapshot(clock[0], **changes)) + assert run.continuity.phase == "waiting" and not run.stop_event.is_set() + run.renew(run.start) + with pytest.raises(StreamSuspended): + run.begin_input(run.start) + clock[0] += 100_000_000 + run.observe_worker(run.start, snapshot(clock[0])) + assert run.continuity.phase == "waiting" + epoch = run.begin_input(run.start) + run.resume_input(epoch, proof(clock[0]), lambda: None) + assert run.continuity.phase == "active" and run.lease.renewals == 1 + + +def test_expired_inventory_can_recover_but_expired_local_lease_cannot(harness): + run, clock = harness(telemetry=True) + for _ in range(4): + clock[0] += 900_000_000 + run.renew(run.start) + assert run.continuity.phase == "waiting" and not run.stop_event.is_set() + run.observe_worker(run.start, snapshot(clock[0])) + epoch = run.begin_input(run.start) + run.resume_input(epoch, proof(clock[0]), lambda: None) + clock[0] += 2_000_000_000 + with pytest.raises(WorkerLeaseError): + run.observe_worker(run.start, snapshot(clock[0])) + assert run.reason == "lease-lost" + + +@pytest.mark.parametrize( + "changes", + [ + {"competing_gpu_clients": ("other-model",)}, + {"gpu_owner_run_id": "other"}, + {"lease_generation": 2}, + {"image_sha256": "f" * 64}, + ], +) +def test_confirmed_owner_or_identity_conflict_still_stops(harness, changes): + run, clock = harness(telemetry=True) + clock[0] += 100_000_000 + with pytest.raises(WorkerLeaseError): + run.observe_worker(run.start, snapshot(clock[0], **changes)) + assert run.stop_event.is_set() and not run.lease.released + + +def test_serial_gpu_lane_survives_a_discarded_epoch(harness): + run, _ = harness() + seen = threading.Event() + + def compute(packet): + if packet["sequence"] == 0: + seen.set() + raise StreamSuspended("discard this frame") + return packet["sequence"] + + stage = SerialGpuStage(run.mailbox, compute, run.stop_event) + try: + run.admit(run.start, {"sequence": 0, "payload_bytes": 8}) + assert seen.wait(1) + run.admit(run.start, {"sequence": 1, "payload_bytes": 8}) + packet, result = stage.take() + assert result == 1 and stage.error is None + run.mailbox.release(packet) + finally: + assert stage.close() + + +@pytest.mark.parametrize("fault", ["eof", "idle", "partial", "gap"]) +def test_connection_loss_keeps_runtime_and_reconnects_without_thread_growth(harness, fault): + run, clock = harness() + epoch = run.start + for index in range(10): + left, right = socket.socketpair() + seen = [] + recv = StreamingIngress( + right, + run, + "capture", + 1, + seen.append, + lambda _: None, + input_epoch=epoch, + idle_timeout=0.05, + fragment_timeout=0.05, + ) + recv.start() + try: + left.sendall(wire.open_packet(epoch, "capture", 1)) + if fault == "eof": + left.shutdown(socket.SHUT_WR) + if fault == "partial": + event = LiveIngressEvent( + 1, "capture", 1, "lidar", "points", index, 0, clock[0], b"points" + ) + header, _ = next(wire.event_packets(epoch, event)) + left.sendall(header + b"p") + left.shutdown(socket.SHUT_WR) + if fault == "gap": + left.sendall( + wire.gap_packet(epoch, modality="camera-frame", reason="source-gap", count=1) + ) + assert recv.join(1) + assert recv.terminal == "paused" and not seen and not run.stop_event.is_set() + assert ( + run.continuity.phase == "waiting" + and run.mailbox.bytes == 0 + and not run.mailbox.done + ) + clock[0] += 1_000_000 + epoch = run.begin_input(run.start) + run.resume_input(epoch, proof(clock[0]), lambda: None) + finally: + left.close() + assert len(run._threads) <= 1 and run.continuity.generation == 10 + + +def test_old_transport_binding_and_backlog_never_reach_new_consumer(harness): + run, clock = harness() + run.pause_input(run.start, "input-timeout") + clock[0] += 100_000_000 + epoch = run.begin_input(run.start) + with pytest.raises(StreamSuspended): + run.check_input(epoch, synchronizing=True, event_ns=clock[0] - 1) + with pytest.raises(StreamSuspended): + run.check_input(run.start, synchronizing=True) + assert not run.stop_event.is_set() + + +def test_late_old_packet_on_new_connection_does_not_kill_models(harness): + run, clock = harness() + run.pause_input(run.start, "input-disconnected") + epoch = run.begin_input(run.start) + run.resume_input(epoch, proof(clock[0]), lambda: None) + left, right = socket.socketpair() + seen = [] + receiver = StreamingIngress( + right, run, "capture", 1, seen.append, lambda _: None, input_epoch=epoch + ) + receiver.start() + try: + left.sendall(wire.open_packet(epoch, "capture", 1)) + left.sendall(wire.terminal_packet(run.start)) + assert receiver.join() and receiver.terminal == "paused" + assert not seen and not run.stop_event.is_set() and not run.mailbox.done + finally: + left.close() diff --git a/tests/test_perception_streaming_decoder.py b/tests/test_perception_streaming_decoder.py index 5bc0c3e..0d3399b 100644 --- a/tests/test_perception_streaming_decoder.py +++ b/tests/test_perception_streaming_decoder.py @@ -162,6 +162,22 @@ def fake_av(monkeypatch): return codec, av, opens +def test_epoch_reset_discards_predictive_state_but_keeps_decoder_object(fake_av): + decoder = decoder_module.FragmentDecoder() + init, frame = fixture() + decoder.configure(init) + decoder.decode(frame) + decoder.reset(init) + assert decoder.frames == 0 and decoder._next_dts is None + decoder.decode(fixture(dts=500)[1]) + assert decoder.frames == 1 and decoder._next_dts == 600 + decoder.reset(init) + with pytest.raises(decoder_module.StreamingDecodeError, match="random-access"): + decoder.decode(fixture(dts=600, sync=False)[1]) + with pytest.raises(decoder_module.StreamingDecodeError, match="failed decoder"): + decoder.reset(init) + + def test_persistent_codec_emits_current_frame_without_demux_or_flush(fake_av): codec, _, opens = fake_av init, fragment = fixture()