test(perception): record cross-host gRPC recovery and backpressure proof
This commit is contained in:
DCCONSTRUCTIONS
@@ -0,0 +1,262 @@
|
||||
"""Bounded CPU-only cross-host transport probe, NOT perception qualification.
|
||||
|
||||
Run server in an isolated Worker Linux container and client on the operator
|
||||
host through the existing SSH tunnel. Ephemeral access files travel via SSH,
|
||||
never stdout/manifests. The source clock and resume proof are SYNTHETIC: this
|
||||
probe measures same-host round trips, not cross-host frame age or model FPS.
|
||||
"""
|
||||
|
||||
import argparse
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import resource
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
import time
|
||||
from datetime import UTC, datetime
|
||||
from hashlib import sha256
|
||||
from pathlib import Path
|
||||
|
||||
import grpc
|
||||
|
||||
from k1link.compute.live_perception import LiveIngressEvent
|
||||
from k1link.perception.realtime_contract import StreamStart
|
||||
from k1link.perception.streaming_continuity import ResumeEvidence
|
||||
from k1link.perception.streaming_grpc import GrpcStreamClient, GrpcStreamEndpoint, StreamAccess
|
||||
from k1link.perception.streaming_lifecycle import StreamingLifecycle
|
||||
from k1link.perception.streaming_queue import StreamMailbox
|
||||
|
||||
|
||||
def document(path, value):
|
||||
path.write_text(json.dumps(value, indent=2) + "\n")
|
||||
|
||||
|
||||
def identity():
|
||||
return StreamStart(
|
||||
"grpc-cpu-canary",
|
||||
"synthetic",
|
||||
"worker-006",
|
||||
"epoch-1",
|
||||
1,
|
||||
*(["a" * 64] * 4),
|
||||
"synthetic-fixed-clock",
|
||||
"live",
|
||||
)
|
||||
|
||||
|
||||
async def until(predicate, seconds=30):
|
||||
async with asyncio.timeout(seconds):
|
||||
while not predicate():
|
||||
await asyncio.sleep(0.02)
|
||||
|
||||
|
||||
async def server(root):
|
||||
private = root / "private"
|
||||
private.mkdir(mode=0o700, exist_ok=True)
|
||||
cert, key = private / "cert.pem", private / "key.pem"
|
||||
subprocess.run(
|
||||
[
|
||||
"openssl",
|
||||
"req",
|
||||
"-x509",
|
||||
"-newkey",
|
||||
"rsa:2048",
|
||||
"-nodes",
|
||||
"-days",
|
||||
"1",
|
||||
"-subj",
|
||||
"/CN=localhost",
|
||||
"-keyout",
|
||||
str(key),
|
||||
"-out",
|
||||
str(cert),
|
||||
],
|
||||
check=True,
|
||||
capture_output=True,
|
||||
timeout=10,
|
||||
)
|
||||
run = StreamingLifecycle(
|
||||
identity(),
|
||||
root / "lease",
|
||||
StreamMailbox(),
|
||||
threading.Event(),
|
||||
recover_input=True,
|
||||
source_clock_ns=lambda: 1_000_000_000,
|
||||
)
|
||||
resident = run.spawn(
|
||||
lambda: subprocess.Popen(
|
||||
[sys.executable, "-c", "import time; time.sleep(100)"], start_new_session=True
|
||||
)
|
||||
)
|
||||
run.ready()
|
||||
pulse_stop = threading.Event()
|
||||
|
||||
def pulse():
|
||||
while not pulse_stop.wait(0.1):
|
||||
run.renew(identity())
|
||||
|
||||
pulse_thread = threading.Thread(target=pulse, daemon=True)
|
||||
pulse_thread.start()
|
||||
received = []
|
||||
|
||||
def consume(event):
|
||||
item = {
|
||||
"sequence": event.ingress_sequence,
|
||||
"bytes": len(event.payload),
|
||||
"sha256": sha256(event.payload).hexdigest(),
|
||||
"resident_pid": resident.pid,
|
||||
}
|
||||
received.append(item)
|
||||
if len(received) > 16:
|
||||
raise ValueError("bounded canary count exceeded")
|
||||
endpoint.publish(
|
||||
run.continuity.epoch, event.ingress_sequence, json.dumps(item, sort_keys=True).encode()
|
||||
)
|
||||
|
||||
endpoint = GrpcStreamEndpoint(run, consume, lambda _: None)
|
||||
rpc = None
|
||||
report = {
|
||||
"schema_version": "missioncore.grpc-cpu-probe/v1",
|
||||
"utc": datetime.now(UTC).isoformat(),
|
||||
"started_monotonic_ns": time.monotonic_ns(),
|
||||
"grpc_version": grpc.__version__,
|
||||
"synthetic_source_clock_and_resume_proof": True,
|
||||
"model_count": 0,
|
||||
"actuation_allowed": False,
|
||||
"real_time_qualified": False,
|
||||
"received": received,
|
||||
}
|
||||
|
||||
def grant(number):
|
||||
access = endpoint.issue(run.continuity.epoch, "synthetic-capture", 1)
|
||||
target = private / f"access-{number}.json"
|
||||
document(
|
||||
target,
|
||||
{
|
||||
"epoch": access.epoch.to_dict(),
|
||||
"session_id": access.session_id,
|
||||
"session_generation": access.session_generation,
|
||||
"token": access.token,
|
||||
},
|
||||
)
|
||||
target.chmod(0o600)
|
||||
|
||||
try:
|
||||
rpc, _ = await endpoint.serve(
|
||||
"0.0.0.0:50061", certificate=cert.read_bytes(), private_key=key.read_bytes()
|
||||
)
|
||||
grant(1)
|
||||
await until(lambda: len(received) == 8 and endpoint.active is None)
|
||||
report["first_transport"] = endpoint.last
|
||||
report["waiting_before"] = run.snapshot()
|
||||
report["pid_before_gap"] = resident.pid
|
||||
began = time.monotonic_ns()
|
||||
await asyncio.sleep(2.2)
|
||||
report["gap_ns"] = time.monotonic_ns() - began
|
||||
assert resident.poll() is None and run.continuity.phase == "waiting"
|
||||
report["waiting_after"] = run.snapshot()
|
||||
epoch = run.begin_input(identity())
|
||||
# Deliberately synthetic fixture proof; the real graph's decoder/sensor
|
||||
# adapter supplies this proof in production, NEVER the network client.
|
||||
run.resume_input(epoch, ResumeEvidence(*([1_000_000_000] * 4), True), lambda: None)
|
||||
grant(2)
|
||||
await until(lambda: len(received) == 16 and run.mailbox.done)
|
||||
endpoint.finish_results(epoch)
|
||||
await until(lambda: endpoint.active is None)
|
||||
report["second_transport"] = endpoint.last
|
||||
report["final_active"] = run.snapshot()
|
||||
report["pid_after_gap"] = resident.pid
|
||||
report["peak_mailbox_bytes"] = run.mailbox.peak_bytes
|
||||
report["rss_max_kib"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
|
||||
report["passed"] = resident.poll() is None
|
||||
finally:
|
||||
if rpc:
|
||||
await rpc.stop(0)
|
||||
pulse_stop.set()
|
||||
pulse_thread.join(1)
|
||||
report["closed"] = run.close()
|
||||
report["final_mailbox_bytes"] = run.mailbox.bytes
|
||||
report["resident_reaped"] = resident.poll() is not None
|
||||
report["finished_monotonic_ns"] = time.monotonic_ns()
|
||||
document(root / "server-report.json", report)
|
||||
for path in (key, private / "access-1.json", private / "access-2.json"):
|
||||
path.unlink(missing_ok=True)
|
||||
|
||||
|
||||
async def client(root, phase, target):
|
||||
access_doc = json.loads((root / f"access-{phase}.json").read_text())
|
||||
access = StreamAccess(
|
||||
StreamStart.from_dict(access_doc["epoch"]),
|
||||
access_doc["session_id"],
|
||||
access_doc["session_generation"],
|
||||
access_doc["token"],
|
||||
)
|
||||
stream = GrpcStreamClient(target, (root / "cert.pem").read_bytes(), access)
|
||||
observations = []
|
||||
report = {
|
||||
"utc": datetime.now(UTC).isoformat(),
|
||||
"started_monotonic_ns": time.monotonic_ns(),
|
||||
"phase": phase,
|
||||
"grpc_version": grpc.__version__,
|
||||
"observations": observations,
|
||||
"source_duration_known": False,
|
||||
"model_count": 0,
|
||||
"path": "TLS gRPC over existing authenticated SSH tunnel",
|
||||
"one_way_age_measured": False,
|
||||
}
|
||||
try:
|
||||
await stream.open()
|
||||
for sequence in range((phase - 1) * 8 + 1, phase * 8 + 1):
|
||||
# One >1 MiB event verifies two wire fragments. All other events are
|
||||
# 64 KiB; this is a small transport sample, not Mac load generation.
|
||||
payload = bytes([sequence]) * (1024 * 1024 + 19 if sequence == 1 else 65536)
|
||||
event = LiveIngressEvent(
|
||||
sequence,
|
||||
"synthetic-capture",
|
||||
1,
|
||||
"lidar",
|
||||
"synthetic-lidar",
|
||||
sequence,
|
||||
1_799_999_999_123_456_789,
|
||||
1_000_000_000,
|
||||
payload,
|
||||
)
|
||||
started = time.monotonic_ns()
|
||||
await stream.send(event)
|
||||
response = await stream.receive()
|
||||
returned = json.loads(response[1])
|
||||
assert response[0] == sequence and returned["sha256"] == sha256(payload).hexdigest()
|
||||
assert returned["bytes"] == len(payload)
|
||||
observations.append(
|
||||
{
|
||||
"sequence": sequence,
|
||||
"bytes": len(payload),
|
||||
"sha256": returned["sha256"],
|
||||
"round_trip_ms": (time.monotonic_ns() - started) / 1e6,
|
||||
}
|
||||
)
|
||||
await asyncio.sleep(0.08)
|
||||
if phase == 2:
|
||||
await stream.end()
|
||||
assert await stream.receive() is None
|
||||
report["passed"] = True
|
||||
finally:
|
||||
await stream.close()
|
||||
report["finished_monotonic_ns"] = time.monotonic_ns()
|
||||
document(root / f"client-{phase}-report.json", report)
|
||||
(root / f"access-{phase}.json").unlink(missing_ok=True)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("mode", choices=("server", "client"))
|
||||
parser.add_argument("root", type=Path)
|
||||
parser.add_argument("--phase", type=int, choices=(1, 2), default=1)
|
||||
parser.add_argument("--target", default="localhost:18561")
|
||||
args = parser.parse_args()
|
||||
os.umask(0o077)
|
||||
asyncio.run(
|
||||
server(args.root) if args.mode == "server" else client(args.root, args.phase, args.target)
|
||||
)
|
||||
Reference in New Issue
Block a user