258 lines
8.7 KiB
Python
258 lines
8.7 KiB
Python
"""Small synthetic ownership/lifecycle checks; no model or GPU workload."""
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
from dataclasses import replace
|
|
|
|
import pytest
|
|
|
|
from k1link.perception.graph_contracts import GraphState
|
|
from k1link.perception.realtime_contract import StreamStart
|
|
from k1link.perception.streaming_lifecycle import StreamingLifecycle
|
|
from k1link.perception.streaming_queue import StreamMailbox
|
|
from k1link.perception.worker_lease import WorkerLeaseError
|
|
|
|
|
|
def start(generation=1):
|
|
return StreamStart(
|
|
run_id=f"run-{generation}",
|
|
source_id="fixture",
|
|
worker_id="worker-006",
|
|
epoch_id=f"epoch-{generation}",
|
|
lease_generation=generation,
|
|
profile_sha256="a" * 64,
|
|
image_sha256="b" * 64,
|
|
effective_config_sha256="c" * 64,
|
|
calibration_sha256="d" * 64,
|
|
clock_domain_id="source-clock",
|
|
input_mode="live",
|
|
)
|
|
|
|
|
|
def runtime(tmp_path, *, generation=1, clock=None):
|
|
return StreamingLifecycle(
|
|
start(generation),
|
|
tmp_path,
|
|
StreamMailbox(),
|
|
threading.Event(),
|
|
**({"clock_ns": clock} if clock is not None else {}),
|
|
)
|
|
|
|
|
|
def test_only_ready_admits_and_clean_stop_requires_new_generation_and_epoch(tmp_path):
|
|
run = runtime(tmp_path)
|
|
packet = {"sequence": 0, "payload_bytes": 20}
|
|
try:
|
|
assert run.state == GraphState.STARTING
|
|
with pytest.raises(WorkerLeaseError, match="accepting"):
|
|
run.admit(run.start, packet)
|
|
run.ready()
|
|
assert run.admit(run.start, packet)
|
|
assert run.mailbox.take() is packet
|
|
with (
|
|
run.work(run.start, "gpu"),
|
|
pytest.raises(WorkerLeaseError, match="one active"),
|
|
run.work(run.start, "gpu"),
|
|
):
|
|
pytest.fail("second GPU callback admitted")
|
|
run.validate_result_binding(run.start.to_dict())
|
|
assert not run.close() # The CPU consumer still owns the input.
|
|
assert not run.lease.released and run.state == GraphState.STOPPING
|
|
with pytest.raises(WorkerLeaseError, match="owned"):
|
|
runtime(tmp_path, generation=2)
|
|
run.mailbox.release(packet)
|
|
finally:
|
|
assert run.close()
|
|
assert run.state == GraphState.STOPPED
|
|
with pytest.raises(WorkerLeaseError, match="generation"):
|
|
runtime(tmp_path)
|
|
next_run = runtime(tmp_path, generation=2)
|
|
assert next_run.close()
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"field,value",
|
|
[
|
|
("epoch_id", "old"),
|
|
("run_id", "old"),
|
|
("worker_id", "other-worker"),
|
|
("source_id", "other-source"),
|
|
("profile_sha256", "e" * 64),
|
|
("image_sha256", "e" * 64),
|
|
("effective_config_sha256", "e" * 64),
|
|
("calibration_sha256", "e" * 64),
|
|
("clock_domain_id", "other-clock"),
|
|
("input_mode", "recorded-source-paced"),
|
|
("lease_generation", 2),
|
|
],
|
|
)
|
|
def test_stale_client_cannot_compute_publish_or_cancel_current_owner(tmp_path, field, value):
|
|
run = runtime(tmp_path)
|
|
try:
|
|
run.ready()
|
|
old = replace(run.start, **{field: value})
|
|
with pytest.raises(WorkerLeaseError, match="mismatch"):
|
|
run.renew(old)
|
|
with pytest.raises(WorkerLeaseError, match="mismatch"):
|
|
run.admit(old, {"sequence": 0, "payload_bytes": 1})
|
|
with pytest.raises(WorkerLeaseError, match="mismatch"):
|
|
run.validate_result_binding(old.to_dict())
|
|
assert not run.stop_event.is_set() and run.state == GraphState.RUNNING
|
|
run.check_current(run.start)
|
|
finally:
|
|
assert run.close()
|
|
|
|
|
|
def test_expiry_cannot_be_renewed_and_never_releases_an_active_callback(tmp_path):
|
|
clock = [1_000_000_000]
|
|
run = runtime(tmp_path, clock=lambda: clock[0])
|
|
run.ready()
|
|
try:
|
|
with pytest.raises(WorkerLeaseError), run.work(run.start, "gpu"):
|
|
clock[0] += 2_000_000_000
|
|
with pytest.raises(WorkerLeaseError):
|
|
run.renew(run.start)
|
|
assert run.stop_event.is_set()
|
|
assert not run.close()
|
|
with pytest.raises(WorkerLeaseError, match="owned"):
|
|
runtime(tmp_path, generation=2)
|
|
assert run.state == GraphState.STOPPING and run.reason == "lease-lost"
|
|
finally:
|
|
assert run.close()
|
|
assert run.state == GraphState.FAILED
|
|
|
|
|
|
def test_heartbeat_uses_receipt_time_and_backwards_clock_fences(tmp_path):
|
|
clock = [1_000_000_000]
|
|
run = runtime(tmp_path, clock=lambda: clock[0])
|
|
try:
|
|
clock[0] += 1_900_000_000
|
|
run.renew(run.start)
|
|
clock[0] += 1_900_000_000
|
|
run.ready()
|
|
clock[0] -= 1
|
|
with pytest.raises(WorkerLeaseError):
|
|
run.check_current(run.start)
|
|
assert run.reason == "lease-lost"
|
|
finally:
|
|
assert run.close()
|
|
|
|
|
|
def test_watchdog_stops_a_real_owned_child_without_waiting_for_another_frame(tmp_path):
|
|
clock = [1_000_000_000]
|
|
run = runtime(tmp_path, clock=lambda: clock[0])
|
|
child = run.spawn(
|
|
lambda: subprocess.Popen(
|
|
[sys.executable, "-c", "import time; print('ready', flush=True); time.sleep(10)"],
|
|
stdout=subprocess.PIPE,
|
|
start_new_session=True,
|
|
)
|
|
)
|
|
try:
|
|
assert child.stdout.readline() == b"ready\n"
|
|
run.ready()
|
|
clock[0] += 2_000_000_000
|
|
assert run.stop_event.wait(1)
|
|
child.wait(timeout=2)
|
|
with pytest.raises(WorkerLeaseError):
|
|
run.validate_result_binding(run.start.to_dict())
|
|
assert not run.lease.released # Expiry is not retirement.
|
|
finally:
|
|
assert run.close()
|
|
child.stdout.close()
|
|
|
|
|
|
def test_provider_failure_fences_new_work_and_pending_input(tmp_path):
|
|
run = runtime(tmp_path)
|
|
run.ready()
|
|
run.admit(run.start, {"sequence": 0, "payload_bytes": 1})
|
|
try:
|
|
with pytest.raises(ValueError, match="provider"), run.work(run.start, "cpu"):
|
|
raise ValueError("provider failed")
|
|
assert run.mailbox.quiescent and run.reason == "failed"
|
|
with pytest.raises(WorkerLeaseError):
|
|
run.check_current(run.start)
|
|
finally:
|
|
assert run.close()
|
|
|
|
|
|
def test_live_registered_thread_prevents_retirement(tmp_path):
|
|
run = runtime(tmp_path)
|
|
release = threading.Event()
|
|
thread = threading.Thread(target=release.wait)
|
|
run.track_thread(thread)
|
|
thread.start()
|
|
try:
|
|
assert not run.close()
|
|
assert not run.lease.released
|
|
finally:
|
|
release.set()
|
|
thread.join(timeout=1)
|
|
assert run.close()
|
|
|
|
|
|
def test_crashed_controller_remains_quarantined_after_os_lock_is_gone(tmp_path):
|
|
program = """
|
|
import json, os, sys
|
|
from pathlib import Path
|
|
from k1link.perception.worker_lease import WorkerLease
|
|
from k1link.perception.realtime_contract import StreamStart
|
|
lease = WorkerLease(Path(sys.argv[1]), StreamStart.from_dict(json.loads(sys.argv[2])))
|
|
os._exit(17)
|
|
"""
|
|
child = subprocess.run(
|
|
[sys.executable, "-c", program, str(tmp_path), json.dumps(start().to_dict())],
|
|
timeout=3,
|
|
capture_output=True,
|
|
)
|
|
assert child.returncode == 17, child.stderr.decode()
|
|
with pytest.raises(WorkerLeaseError, match="unretired"):
|
|
runtime(tmp_path, generation=2)
|
|
assert json.loads((tmp_path / "owner.json").read_text())["state"] == "active"
|
|
|
|
|
|
def test_cross_process_contender_is_denied_without_mutating_owner_record(tmp_path):
|
|
run = runtime(tmp_path)
|
|
before = (tmp_path / "owner.json").read_bytes()
|
|
program = """
|
|
import json, sys
|
|
from pathlib import Path
|
|
from k1link.perception.worker_lease import WorkerLease, WorkerLeaseError
|
|
from k1link.perception.realtime_contract import StreamStart
|
|
try:
|
|
WorkerLease(Path(sys.argv[1]), StreamStart.from_dict(json.loads(sys.argv[2])))
|
|
except WorkerLeaseError as exc:
|
|
print(str(exc)); sys.exit(21)
|
|
sys.exit(99)
|
|
"""
|
|
try:
|
|
child = subprocess.run(
|
|
[sys.executable, "-c", program, str(tmp_path), json.dumps(start(2).to_dict())],
|
|
capture_output=True,
|
|
timeout=3,
|
|
)
|
|
assert child.returncode == 21 and b"already owned" in child.stdout
|
|
assert (tmp_path / "owner.json").read_bytes() == before
|
|
finally:
|
|
assert run.close()
|
|
|
|
|
|
def test_unknown_or_replaced_lock_cannot_grant_authority(tmp_path):
|
|
run = runtime(tmp_path)
|
|
run.ready()
|
|
original = tmp_path / "original.lock"
|
|
os.rename(tmp_path / "worker.lock", original)
|
|
(tmp_path / "worker.lock").write_bytes(b"")
|
|
try:
|
|
with pytest.raises(WorkerLeaseError):
|
|
run.check_current(run.start)
|
|
with pytest.raises(WorkerLeaseError, match="identity"):
|
|
run.close()
|
|
finally:
|
|
# Synthetic fixture only: restore exact inode so cleanup can retire.
|
|
os.replace(original, tmp_path / "worker.lock")
|
|
assert run.close()
|