354 lines
12 KiB
Python
354 lines
12 KiB
Python
"""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(StreamSuspended, match="still active"):
|
|
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()
|