From 191612282bd1ed51f61ef56bafdaf562f3bca77d Mon Sep 17 00:00:00 2001 From: DCCONSTRUCTIONS Date: Wed, 2 Sep 2026 18:14:51 +0300 Subject: [PATCH] test(perception): record cross-host gRPC recovery and backpressure proof --- .../OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md | 8 +- .../0049-stream-first-perception-profiles.md | 66 ++++- .../PERCEPTION_STREAM_STAGE1_2026-09-01.md | 45 +++ .../perception/worker/grpc_transport_probe.py | 262 ++++++++++++++++++ 4 files changed, 376 insertions(+), 5 deletions(-) create mode 100644 experiments/perception/worker/grpc_transport_probe.py diff --git a/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md b/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md index ed26c20..fe0c07c 100644 --- a/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md +++ b/docs/OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md @@ -1,6 +1,12 @@ # Observatory: четыре этапа создания полного real-time Perception-профиля -Дата: 2026-09-01; обновлено 2026-09-02 17:37 МСК. **Этап 1 закрыт; этап 2 в работе. Инкремент 11: восстановлен существующий System/Worker telemetry plane и добавлен независимый экспорт состояния streaming runtime.** Код `c570f7d`, `2ef8c08`. Исправлены остановка Telegraf при недоступном MQTT, потерянная публикация порта, превышающие лимит контейнера настройки PostgreSQL и блокирующая ingest очистка истории. CPU-only переходы running → waiting → synchronizing → running прошли через реальный агент/Timescale. Это observability, не новый внешний sensor data-plane; следующий пункт остаётся binary stream/auth/recovery между приложением и Worker, затем standalone packaging. Этапы3–4 не начаты. +Дата: 2026-09-01; обновлено 2026-09-02 18:11 МСК. **Этап 1 закрыт; этап 2 в работе. Инкремент 12: внешний gRPC/TLS binary candidate, одноразовый доступ и bounded backpressure.** Код `0310592`. Два CPU-only межмашинных повтора Mac↔Worker через существующий SSH-маршрут дали16/16 точных payload/reply; обрыв2.20s сохранил resident CPU sentinel и локальный lease. Итоговые105 focused tests и20 Worker transport tests PASS. Это не full-graph network qualification: реальный `BinaryGraphBridge`, scene/layer codecs, production выдача grant/clock mapping и standalone ещё не подключены к этому endpoint. Этапы3–4 не начаты. + +Финальный same-Mac RTT min/median/max14.360/23.919/120.815ms (не one-way age/FPS), peak mailbox6,750,559B, RSS57,244KiB, после закрытия0bytes. Source clock и resume proof синтетические, моделей0. Default64KiB HTTP/2 window не выдержал250ms для1,048,595B; fixed1MiB прошёл без изменения deadline. Реальный slow-reader выявил блокирование отправки error status; его deadline и deferred quarantine cleanup исправлены и проверены. Evidence16 artifacts: `.runtime/perception-stage2-grpc-20260902T1500Z/manifest.json`, SHA-256 `adf5eaf092feaed6721f66e2adaceded0cdbf55754e1f9953f54623bfb52d331`. + +Следующий проверяемый результат этапа2: подать реальную запись с application-side source через этот endpoint в существующий полный graph, вернуть общие scene/layer outputs с явными размерными/возрастными ограничениями, повторить normal/reconnect/slow-consumer canary последовательно. Транспорт не имеет права вызывать batch materializer или заново запускать модели при сетевой паузе. Текущий outbound cap1MiB требует проверки реальных слоёв/фрагментации; расширять его молча нельзя. После этого — самостоятельный image. Продуктовый LAB path пока не переключён. + +Предыдущий инкремент11 (`c570f7d`, `2ef8c08`): восстановлены System/Worker telemetry plane и независимый runtime exporter. Телеметрия остаётся отдельным наблюдателем, а не выдающим lease/GPU authority контуром. 104 Python tests, 703 frontend tests, typecheck и production build PASS. Полный frontend проверен последовательно на CPU Worker из-за Mac memory pressure=2; собранный UI установлен на canonical8000. Визуальная приёмка по `mission-core-product-ui` отложена до освобождения памяти, не объявлена пройденной. Расширен существующий Worker, без новых страниц и без product profile/registry cutover. MQTT outage/startup проверен с сохранением PID9268; CPU canary сохранил lease generation1 и освободил его после завершения. Evidence `.runtime/worker-telemetry-20260902T1422Z/manifest.json`, SHA-256 `825b1d8035dbf907b2b731d35ede4c5742b81e653a6f24670f4775981f853492`. diff --git a/docs/adr/0049-stream-first-perception-profiles.md b/docs/adr/0049-stream-first-perception-profiles.md index 236b340..b5d0fb2 100644 --- a/docs/adr/0049-stream-first-perception-profiles.md +++ b/docs/adr/0049-stream-first-perception-profiles.md @@ -1,9 +1,9 @@ # ADR 0049 — Stream-first full perception profiles -Date: 2026-09-01; updated 2026-09-02 16:52 MSK. Status: stage-1 engineering -baseline complete; stage-2 full binary GPU pilot consumes real scoped host facts. -External transport/authentication, native-host GPU inventory, standalone image -and product cutover remain open; local proof is not physical-live acceptance. +Date: 2026-09-01; updated 2026-09-02 18:11 MSK. Status: stage-1 engineering +baseline complete; stage-2 authenticated gRPC candidate has cross-host CPU proof. +Full-graph network integration, native-host GPU inventory, standalone image and +product cutover remain open; transport proof is not physical-live acceptance. Execution order remains the four stages in [the ExecPlan](../OBSERVATORY_REALTIME_PROFILES_EXECPLAN.md). @@ -719,3 +719,61 @@ integration does not constitute the external camera/points binary data plane. See `experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md` for measured component results, bounded validation, Worker maintenance and remaining gates. Neither this ADR nor passing synthetic contract tests marks the new runtime ready. + +## Stage 2 increment 12: external gRPC/TLS candidate (2026-09-02) + +`streaming_grpc.py` exposes `/missioncore.perception.v1.BinaryStream/Exchange`. +It is an optional `perception-stream` dependency, locked grpcio1.83.1; it is not +auto-started by the backend or installed into the product LAB path. Both sides +require TLS; the client verifies its supplied trust root and server name. The +local controller alone issues a256-bit random single-use capability with30s +admission expiry, bound to the complete StreamStart and acquisition session. +The endpoint retains only its digest. Invalid/duplicate/stale credentials cannot +reserve a stream, stop its owner, renew a lease, launch a model or create an epoch. +One active stream is admitted; reconnect needs a controller-approved new epoch +and, separately, real decoder/causal-sensor resumption proof. Network disconnect +is not explicit End or Cancel. Production grant delivery is still an integration +gate; the CPU probe transfers ephemeral files only through authenticated SSH. + +The generic gRPC method uses identity byte serialization, not generated protobuf +messages. Request messages contain≤64KiB pieces of the unchanged MCI2 wire; +StreamingIngress still validates metadata, fragments, hashes and epoch/source +binding. No archive, duration, file path or executable command is sent. Response +MCR3 is only an envelope over existing domain bytes: magic4, uint64 sequence8, +binding SHA25632, payload SHA25632, then1..1MiB payload. It does not invent a new +scene ontology or claim that full-graph scene/layer serialization is connected. +The actual profile's larger outputs require reviewed fragmentation, not a silent +cap increase. Result age validation and the full result receipt ledger remain +with the graph/application adapter, not the transport envelope. + +There are two pending latest replies and one writer/reader per direction; no +background source queue/retry or event-loop callback backlog. Python bridge +storage is reserved inside the existing16MiB mailbox budget. Fixed1MiB HTTP/2 +lookahead, disabled BDP growth/retries, bounded metadata/messages/IO are explicit; +native gRPC/TLS/socket memory also needs measured RSS, not a claimed byte-perfect +mapping into mailbox accounting. Public/multi-client DoS hardening is not proved +by this private Worker experiment. A250ms event deadline covers the complete +source event. Even final error-status sending is deadline-bound; otherwise an +unread response can delay cleanup. Late trusted callbacks are quarantined, with +no replacement grant or early byte release until they return. Pending reply +drops are counted; they are not equivalent to end-to-end delivery receipts. + +Evidence: initial default64KiB lookahead timed out on1,048,595B. Fixed1MiB passed +two16/16 cross-host samples without changing250ms. Final min/median/max RTT +14.360/23.919/120.815ms on the Mac includes SSH/echo overhead, not clock-subtracted +one-way age.2.202s wait retained the same CPU sentinel PID40 and lease; renewals +32→54. Source clock and resume evidence are explicitly synthetic, model count0. +Peak mailbox6,750,559B; Linux RSS57,244KiB.20 Worker transport tests include1MiB +output, a real slow reader and late-callback cleanup;105 focused Mac tests pass +with the two larger probes intentionally skipped. Final cleanup leaves0bytes, +no temporary listener/container/tunnel/private key/grant. No GPU setters or +durable services changed. Frigate/Ollama remain disabled, Telegraf is Running. +Manifest SHA256 `adf5eaf092feaed6721f66e2adaceded0cdbf55754e1f9953f54623bfb52d331` +at `.runtime/perception-stage2-grpc-20260902T1500Z/manifest.json` (16 artifacts). + +Design references: [gRPC flow control](https://grpc.io/docs/guides/flow-control/) +warns that a completed write need not mean network delivery; +[authentication](https://grpc.io/docs/guides/auth/) documents TLS credentials; +[channel options](https://github.com/grpc/grpc/blob/master/include/grpc/impl/channel_arg_names.h) +documents the fixed lookahead window. Source and actual behavior were checked; +these references alone are not a performance qualification. diff --git a/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md b/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md index ec7a1c6..9ec0ddb 100644 --- a/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md +++ b/experiments/perception/PERCEPTION_STREAM_STAGE1_2026-09-01.md @@ -2303,3 +2303,48 @@ Telegraf/telemetry plane healthy. Frigate/Ollama exited/restart=no. Final GPU: Этап2 остаётся открыт: после визуального acceptance продолжить внешний bounded binary data-plane/auth/recovery, затем standalone image. Этапы3–4 не начаты; расширение System/Worker не означает интеграцию нового профиля в Observatory. + +## Stage 2 increment 12 — gRPC/TLS cross-host CPU transport, 2026-09-02 18:11 MSK + +Код `0310592`: optional `perception-stream` extra (locked grpcio1.83.1), +`streaming_grpc.py` и `streaming_network.py`. Существующий MCI2 raw ingress идёт +через TLS gRPC pieces≤64KiB; ответы≤1MiB обёрнуты binding/sequence/SHA256, не +пересобирают scene ontology. Контроллер выдаёт single-use256-bit grant с30s +admission TTL; сеть не может получить lease, запустить модели или разрешить +resume. Два pending latest replies,250ms event/IO deadlines, bounded metadata, +fixed1MiB HTTP/2 lookahead, retries/BDP growth disabled. Приложение8000 и +полный граф ещё не подключены к этому endpoint. + +Probe `worker/grpc_transport_probe.py` использует one CPU/512MiB, runtime=runc, +no NVIDIA/GPU, один CPU sentinel. Локальный lifecycle heartbeat независим от +TLS/SSH соединения. Source clock/resume proof специально синтетические; это +transport test, не замер source-paced модели и не проверка decoder resync. + +Первый default64KiB lookahead не передал1,048,595B в250ms;0observations completed, +cleanup успешен. После fixed1MiB два16/16 exact повтора без расширения deadline. +Последний на финальном snapshot017f7871…: RTT14.360/23.919/120.815ms +min/median/max,2.202367178s WAIT, CPU PID40 сохранён, renewals32→54, generation1 +lease и новый input epoch. Peak input/bridge6,750,559B, RSS57,244KiB, final0bytes, +resident reaped. Это same-Mac round-trip через SSH/TLS с echo, не one-way age, +полный scene output или physical uplink300–500Mbps. + +Real slow-reader Worker regression выявил блокировку самой отправки gRPC error +status; теперь и она ограничена deadline. Late trusted callback quarantined до +возврата, replacement denied, отложенное освобождение проверено. Все20 network +tests PASS на Worker,105 focused PASS на Mac;2 крупных probes запускаются только +на Worker.1MiB output round-trip и actual unread-result flow-control проверены. +Ruff/format и strict mypy двух модулей PASS. Initial Worker fixture не имел +typing_extensions; зависимости исправлены в mounted experiment directory, +не в host/global Python. Никакие модели/clocks/durable services не менялись. + +Evidence16 artifacts: `.runtime/perception-stage2-grpc-20260902T1500Z/manifest.json`, +SHA256 `adf5eaf092feaed6721f66e2adaceded0cdbf55754e1f9953f54623bfb52d331`. +Final source hashes Worker/Mac совпали. Temporary containers/tunnels/listeners +и ephemeral private credentials отсутствуют.982MiB/0%GPU/auto210/405MHz/450W; +Telegraf Running, Frigate/Ollama exited/restart=no, canonical8000 здоров. +UI/build/браузер не запускались; final Mac memory pressure2. + +Следующий gate: real BinaryGraphBridge/source/scene codecs на внешнем канале, +production grant delivery и clock mapping, layer-size/freshness/result receipt +accounting, serial full-graph canary; затем standalone. Не переключать общий +LAB path и не объявлять real-time acceptance по этим CPU цифрам. diff --git a/experiments/perception/worker/grpc_transport_probe.py b/experiments/perception/worker/grpc_transport_probe.py new file mode 100644 index 0000000..f84e818 --- /dev/null +++ b/experiments/perception/worker/grpc_transport_probe.py @@ -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) + )