From a1ed8d8fc53f9f4c279278718f0568a8f15c0eab Mon Sep 17 00:00:00 2001 From: Mikkeli Date: Fri, 9 Oct 2026 20:19:05 +0900 Subject: [PATCH] Boundary v0: WebSocket with heartbeats, acked commands and mock state One socket on :7700 carries hello, heartbeats both ways (5 s timeout), 5 Hz state snapshots of envelopes, and idempotent commands. 'mark' is forwarded to the recorder's control socket. --mock synthesises degrading state flagged as mock in every envelope; without it a client gets hello and heartbeats only. Peers outside loopback, RFC 1918 and the tailnet are refused. Deployed on the car Orin as yukart-gw.service. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz --- Cargo.lock | 298 +++++++++++++++++++++++++++++++++++++ Cargo.toml | 6 + README.md | 7 + deploy/yukart-gw.service | 14 ++ docs/impl/plans/INITIAL.md | 20 ++- scripts/deploy.sh | 31 ++++ src/client.rs | 141 ++++++++++++++++++ src/hub.rs | 58 ++++++++ src/main.rs | 133 ++++++++++++++++- src/mock.rs | 59 ++++++++ src/signals.rs | 45 ++++++ 11 files changed, 809 insertions(+), 3 deletions(-) create mode 100644 deploy/yukart-gw.service create mode 100755 scripts/deploy.sh create mode 100644 src/client.rs create mode 100644 src/hub.rs create mode 100644 src/mock.rs create mode 100644 src/signals.rs diff --git a/Cargo.lock b/Cargo.lock index ec3c3a5..372bf8e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2,6 +2,304 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "block-buffer" +version = "0.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d2f6c7dbe95a6ed67ad9f18e57daf93a2f034c524b99fd2b76d18fdfeb6660aa" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "bytes" +version = "1.12.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc652a48c352aef3ea3aed32080501cf3ef6ed5da78602a020c991775b0aff04" + +[[package]] +name = "cfg-if" +version = "1.0.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4e7648175b45a9a48536d676f68d918270699102aa8dab5496df06904c914600" + +[[package]] +name = "chacha20" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c35e4b699c7e15ccbe7ee35c005e4fc0a278d22238a2857e6ce2dadeda1b06" +dependencies = [ + "cfg-if", + "cpufeatures", + "rand_core", +] + +[[package]] +name = "const-oid" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a6ef517f0926dd24a1582492c791b6a4818a4d94e789a334894aa15b0d12f55c" + +[[package]] +name = "cpufeatures" +version = "0.3.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5ca28b0ae3115b884660db4118d803791fd6756b6e88f39c0f3f7859060d7566" +dependencies = [ + "libc", +] + +[[package]] +name = "crypto-common" +version = "0.2.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce6e4c961d6cd6c9a86db418387425e8bdeaf05b3c8bc1411e6dca4c252f1453" +dependencies = [ + "hybrid-array", +] + +[[package]] +name = "data-encoding" +version = "2.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4583a4551df46e2792f82ceeac45e850d2e2d5debba0b91f102385cda5b11f06" + +[[package]] +name = "digest" +version = "0.11.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f1dd6dbb5841937940781866fa1281a1ff7bd3bf827091440879f9994983d5c2" +dependencies = [ + "block-buffer", + "const-oid", + "crypto-common", +] + +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "rand_core", +] + +[[package]] +name = "http" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "918d3568bebf352712bc2ef3d46a8bcf1a75b373be6539de198e9105cbbf9ce0" +dependencies = [ + "bytes", + "itoa", +] + +[[package]] +name = "httparse" +version = "1.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6dbf3de79e51f3d586ab4cb9d5c3e2c14aa28ed23d180cf89b4df0454a69cc87" + +[[package]] +name = "hybrid-array" +version = "0.4.15" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "27f864f10dfb56725ce5ce5472bc52252c8f93a4ab86327122cebf62c5f59a17" +dependencies = [ + "typenum", +] + +[[package]] +name = "itoa" +version = "1.0.18" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682" + +[[package]] +name = "libc" +version = "0.2.190" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" + +[[package]] +name = "log" +version = "0.4.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6" + +[[package]] +name = "memchr" +version = "2.8.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" + +[[package]] +name = "proc-macro2" +version = "1.0.107" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "985e7ec9bb745e6ce6535b544d84d6cd6f7ad8bd711c398938ae983b91a766d9" +dependencies = [ + "unicode-ident", +] + +[[package]] +name = "quote" +version = "1.0.47" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fbf4db142a473a8d80c26bbf18454ed458bf8d26c8219c331daecfdbd079001" +dependencies = [ + "proc-macro2", +] + +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + +[[package]] +name = "rand" +version = "0.10.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "65c9fb96cbc91e3478eaae79a69fcd3f1ae4ad052e471fe6732fff548984b4af" +dependencies = [ + "chacha20", + "getrandom", + "rand_core", +] + +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + +[[package]] +name = "serde" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "4148590afebada386688f18773da617792bf2ef03ffc1e4cbd2b1d45b023e0ba" +dependencies = [ + "serde_core", +] + +[[package]] +name = "serde_core" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "67dca2c9c51e58a4791a4b1ed58308b39c64224d349a935ab5039aa360942a48" +dependencies = [ + "serde_derive", +] + +[[package]] +name = "serde_derive" +version = "1.0.229" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e7a5d71263a5a7d47b41f6b3f06ba276f10cc18b0931f1799f710578e2309348" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "serde_json" +version = "1.0.151" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c841b55ecdae098c80dcae9cf767f6f8a0c2cdb3416bbef72181df4d0fe73f14" +dependencies = [ + "itoa", + "memchr", + "serde", + "serde_core", + "zmij", +] + +[[package]] +name = "sha1" +version = "0.11.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "aacc4cc499359472b4abe1bf11d0b12e688af9a805fa5e3016f9a386dc2d0214" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "svc-gw" version = "0.1.0" +dependencies = [ + "libc", + "serde_json", + "tungstenite", +] + +[[package]] +name = "syn" +version = "3.0.6" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8593e8e72159ed2257d083c7a454a85cbf854f37a0966d8d483aff8c8a3ebcee" +dependencies = [ + "proc-macro2", + "quote", + "unicode-ident", +] + +[[package]] +name = "thiserror" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "09e52cb86a36cede5cb101bf8908837b3e4c6e5e59fe7fd85c23fb56200d189e" +dependencies = [ + "thiserror-impl", +] + +[[package]] +name = "thiserror-impl" +version = "2.0.21" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fe5197923287db20a58125f0bc85c062f7f2c892de97b18c356f9efb14b28524" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tungstenite" +version = "0.30.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e48ac77174b19c110a50ab2128b24215ac9cb40e0e12e093fb602d175c569d22" +dependencies = [ + "bytes", + "data-encoding", + "http", + "httparse", + "log", + "rand", + "sha1", + "thiserror", +] + +[[package]] +name = "typenum" +version = "1.20.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20" + +[[package]] +name = "unicode-ident" +version = "1.0.26" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" + +[[package]] +name = "zmij" +version = "1.0.23" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "29666d0abbfad1e3dc4dcf6144730dd3a3ab225bbbdac83319345b1b44ccfc1b" diff --git a/Cargo.toml b/Cargo.toml index 63b3277..c7c6e16 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -6,3 +6,9 @@ description = "Telemetry gateway: aggregation and the app boundary" publish = false [dependencies] +libc = "0.2" +serde_json = "1" +tungstenite = { version = "0.30", default-features = false, features = ["handshake"] } + +[profile.release] +strip = true diff --git a/README.md b/README.md index c2dcbee..362eea6 100644 --- a/README.md +++ b/README.md @@ -15,5 +15,12 @@ HTTP API so a display has one address to talk to. It is a bus subscriber. It never touches hardware and never transmits on CAN. +```bash +cargo run -- --mock # on a workstation: mock state, flagged as mock +scripts/deploy.sh # build, glibc check, install yukart-gw on the car Orin +``` + +Needs [contracts](gitea:yukart/contracts) checked out beside this repo. + - For implementation practice guidelines, refer to the [guidelines](docs/GUIDELINES.md); - For implementation plans and history, refer to [them](docs/impl/); diff --git a/deploy/yukart-gw.service b/deploy/yukart-gw.service new file mode 100644 index 0000000..72be14b --- /dev/null +++ b/deploy/yukart-gw.service @@ -0,0 +1,14 @@ +[Unit] +Description=Yukart gateway: the display boundary (svc-gw) +Documentation=gitea:yukart/svc-gw +StartLimitIntervalSec=0 + +[Service] +User=mikkeli +# Add --mock only on a bench unit: every message then says so, and displays must show it. +ExecStart=/usr/local/bin/yukart-gw --listen 0.0.0.0:7700 +Restart=always +RestartSec=2 + +[Install] +WantedBy=multi-user.target diff --git a/docs/impl/plans/INITIAL.md b/docs/impl/plans/INITIAL.md index 738f5a4..7e3defc 100644 --- a/docs/impl/plans/INITIAL.md +++ b/docs/impl/plans/INITIAL.md @@ -1,7 +1,7 @@ --- type: plan status: active -updated: 2026-09-04 +updated: 2026-10-09 --- # Initial plan @@ -20,6 +20,9 @@ directly. ## Order +0. **Boundary v0** — in place: WebSocket on `:7700`, hello, heartbeats both ways, + `mark` forwarded to the recorder's control socket, mock state behind `--mock`. + Protocol in contracts [`docs/BOUNDARY.md`](gitea:yukart/contracts). 1. Static file serving for the basemap — smallest useful thing, unblocks the map. 2. Nav HTTP proxy to [svc-nav](gitea:yukart/svc-nav). 3. Aggregation and rate limiting, once there is real CAN traffic to aggregate. @@ -75,3 +78,18 @@ Two rules constrain it: indistinguishable from real ones, on a display in a moving car, is a safety problem rather than a tidiness one. The mode is off by default and flagged in the stream, not merely in a log line at startup. + +## Boundary v0, as built + +- One WebSocket rather than separate push and request/response transports: the two + channels keep their different guarantees (drop-oldest down, acked-with-idempotency-key + up) on one socket, and the client has one connection to keep alive. +- Threads, not async: a handful of clients at most, and tens of MB matter more + here than connection count. +- **Peers are filtered by address** — loopback, RFC 1918, tailnet 100.64/10. The car + unit has an LTE uplink, and a command endpoint must not answer on it. +- Without `--mock` and with no bus yet, a client sees hello and heartbeats only. + That is the honest state of the car today, and still exercises the link, + reconnection and marking end to end. +- Signal units are read from contracts' `signals.json` at build time from a sibling + checkout (`../contracts`). The build fails without it rather than guess. diff --git a/scripts/deploy.sh b/scripts/deploy.sh new file mode 100755 index 0000000..ff8f8c3 --- /dev/null +++ b/scripts/deploy.sh @@ -0,0 +1,31 @@ +#!/usr/bin/env bash +# Build svc-gw for aarch64, check it against the car's glibc, and install it +# on an Orin as the yukart-gw service. +# +# usage: scripts/deploy.sh [host] (default: mikkeli-orin-nano) +# +# Cross-compiles from the workstation (the documented fallback): the development +# Orin is the preferred native build host. Either way check-glibc.sh gates it. +set -euo pipefail + +HOST="${1:-mikkeli-orin-nano}" +cd "$(dirname "$0")/.." +TARGET=aarch64-unknown-linux-gnu +BIN="target/$TARGET/release/svc-gw" + +cargo build --release --target "$TARGET" +scripts/check-glibc.sh "$BIN" + +# Binary lives under ~/infra per the host repo's guidelines; /usr/local/bin gets a symlink. +ssh "$HOST" 'mkdir -p ~/infra/svc-gw' +scp -q "$BIN" "$HOST:infra/svc-gw/svc-gw.new" +scp -q deploy/yukart-gw.service "$HOST:infra/svc-gw/yukart-gw.service" +ssh "$HOST" 'set -e + mv ~/infra/svc-gw/svc-gw.new ~/infra/svc-gw/svc-gw + sudo ln -sfn "$HOME/infra/svc-gw/svc-gw" /usr/local/bin/yukart-gw + sudo install -m 0644 ~/infra/svc-gw/yukart-gw.service /etc/systemd/system/yukart-gw.service + sudo systemctl daemon-reload + sudo systemctl enable yukart-gw.service >/dev/null + sudo systemctl restart yukart-gw.service + sleep 1 + systemctl --no-pager --lines=5 status yukart-gw.service' diff --git a/src/client.rs b/src/client.rs new file mode 100644 index 0000000..b27ffa8 --- /dev/null +++ b/src/client.rs @@ -0,0 +1,141 @@ +//! One display client: the two boundary channels over one WebSocket. +//! +//! Down is lossy state (drop-oldest outbox). Up is acknowledged commands, made safe +//! to retry across reconnects by an idempotency key. Both sides heartbeat; a client +//! silent for `TIMEOUT` is dropped, because an asleep tablet still looks connected +//! at the USB link and TCP layers. + +use std::collections::VecDeque; +use std::io::{Read, Write}; +use std::net::TcpStream; +use std::os::unix::net::UnixStream; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use serde_json::{json, Value}; +use tungstenite::{Error, Message}; + +use crate::hub::Hub; + +pub const HEARTBEAT: Duration = Duration::from_secs(1); +pub const TIMEOUT: Duration = Duration::from_secs(5); + +pub struct Ctx { + pub hub: Hub, + pub hello: Value, + pub recorder_ctl: String, + /// Acks by command id, shared across connections so a retry after reconnect is answered, not re-run. + pub acks: Mutex>, +} + +pub fn serve(stream: TcpStream, ctx: Arc) { + let peer = stream.peer_addr().map(|a| a.to_string()).unwrap_or_default(); + let _ = stream.set_read_timeout(Some(Duration::from_secs(5))); + let _ = stream.set_write_timeout(Some(Duration::from_secs(2))); + let _ = stream.set_nodelay(true); + let mut ws = match tungstenite::accept(stream) { + Ok(ws) => ws, + Err(e) => { + eprintln!("{peer}: handshake failed: {e}"); + return; + } + }; + let _ = ws.get_ref().set_read_timeout(Some(Duration::from_millis(50))); + if ws.send(Message::text(ctx.hello.to_string())).is_err() { + return; + } + let outbox = ctx.hub.join(); + eprintln!("{peer}: connected ({} clients)", ctx.hub.count()); + + let mut last_seen = Instant::now(); + let reason = loop { + match ws.read() { + Ok(Message::Text(t)) => { + last_seen = Instant::now(); + if let Some(reply) = handle(&ctx, t.as_str()) { + if ws.send(Message::text(reply.to_string())).is_err() { + break "write failed"; + } + } + } + Ok(Message::Close(_)) => break "closed by client", + Ok(_) => last_seen = Instant::now(), + Err(Error::Io(e)) if matches!(e.kind(), std::io::ErrorKind::WouldBlock | std::io::ErrorKind::TimedOut) => {} + Err(_) => break "connection lost", + } + if last_seen.elapsed() > TIMEOUT { + let _ = ws.close(None); + break "heartbeat timeout"; + } + let mut failed = false; + for m in outbox.drain(Duration::ZERO) { + if ws.send(Message::text(m)).is_err() { + failed = true; + break; + } + } + if failed { + break "write failed"; + } + }; + ctx.hub.leave(&outbox); + eprintln!("{peer}: disconnected: {reason} ({} clients)", ctx.hub.count()); +} + +fn handle(ctx: &Ctx, text: &str) -> Option { + let msg: Value = serde_json::from_str(text).ok()?; + match msg["t"].as_str()? { + "hb" => None, + "cmd" => { + let id = msg["id"].as_str().unwrap_or_default().to_string(); + if id.is_empty() { + return Some(json!({ "t": "ack", "id": null, "ok": false, "detail": "cmd needs an id" })); + } + if let Some((_, ack)) = ctx.acks.lock().unwrap().iter().find(|(k, _)| *k == id) { + let mut a = ack.clone(); + a["dup"] = json!(true); + return Some(a); + } + let (ok, detail) = run(ctx, &msg); + let ack = json!({ "t": "ack", "id": id, "ok": ok, "detail": detail }); + let mut acks = ctx.acks.lock().unwrap(); + if acks.len() >= 256 { + acks.pop_front(); + } + acks.push_back((id, ack.clone())); + Some(ack) + } + _ => None, + } +} + +fn run(ctx: &Ctx, msg: &Value) -> (bool, String) { + match msg["cmd"].as_str().unwrap_or_default() { + "ping" => (true, "pong".into()), + "mark" => { + let text = msg["args"]["text"].as_str().unwrap_or_default().trim().replace('\n', " "); + if text.is_empty() { + return (false, "mark needs args.text".into()); + } + // The recorder stamps the mark on arrival; a retry after an outage arrives + // late by `delay_ms`, so that goes into the mark rather than being lost. + let delay = msg["args"]["delay_ms"].as_u64().unwrap_or(0); + let line = if delay > 500 { format!("mark {text} (delay_ms={delay})") } else { format!("mark {text}") }; + match recorder(&ctx.recorder_ctl, &line) { + Ok(r) => (!r.starts_with("error"), r), + Err(e) => (false, format!("recorder unreachable: {e}")), + } + } + other => (false, format!("unknown command '{other}'")), + } +} + +fn recorder(path: &str, line: &str) -> std::io::Result { + let mut s = UnixStream::connect(path)?; + s.set_read_timeout(Some(Duration::from_secs(3)))?; + s.write_all(line.as_bytes())?; + s.shutdown(std::net::Shutdown::Write)?; + let mut out = String::new(); + s.read_to_string(&mut out)?; + Ok(out.trim().to_string()) +} diff --git a/src/hub.rs b/src/hub.rs new file mode 100644 index 0000000..f6b01bd --- /dev/null +++ b/src/hub.rs @@ -0,0 +1,58 @@ +//! Fan-out to clients. Drop-only: each client has a bounded queue that loses its +//! oldest message when full, so a tablet with its screen off never stalls anything. + +use std::collections::VecDeque; +use std::sync::{Arc, Condvar, Mutex}; +use std::time::Duration; + +const QUEUE: usize = 64; + +pub struct Outbox { + q: Mutex>, + cv: Condvar, +} + +impl Outbox { + pub fn push(&self, msg: String) { + let mut q = self.q.lock().unwrap(); + if q.len() >= QUEUE { + q.pop_front(); + } + q.push_back(msg); + self.cv.notify_one(); + } + + /// Everything queued, waiting at most `wait` for something to arrive. + pub fn drain(&self, wait: Duration) -> Vec { + let q = self.q.lock().unwrap(); + let (mut q, _) = self.cv.wait_timeout_while(q, wait, |q| q.is_empty()).unwrap(); + q.drain(..).collect() + } +} + +#[derive(Default, Clone)] +pub struct Hub { + clients: Arc>>>, +} + +impl Hub { + pub fn join(&self) -> Arc { + let o = Arc::new(Outbox { q: Mutex::new(VecDeque::new()), cv: Condvar::new() }); + self.clients.lock().unwrap().push(o.clone()); + o + } + + pub fn leave(&self, o: &Arc) { + self.clients.lock().unwrap().retain(|c| !Arc::ptr_eq(c, o)); + } + + pub fn broadcast(&self, msg: &str) { + for c in self.clients.lock().unwrap().iter() { + c.push(msg.to_string()); + } + } + + pub fn count(&self) -> usize { + self.clients.lock().unwrap().len() + } +} diff --git a/src/main.rs b/src/main.rs index 563cee7..669f8ad 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,3 +1,132 @@ -fn main() { - println!("{} is not implemented yet", env!("CARGO_PKG_NAME")); +//! svc-gw — the boundary to display clients. Protocol v0: one WebSocket carrying +//! hello, heartbeats, state snapshots down and acknowledged commands up. +//! Defined in contracts `docs/BOUNDARY.md`. +//! +//! v0 has no bus input yet: without `--mock` it sends hello and heartbeats only. + +mod client; +mod hub; +mod mock; +mod signals; + +use std::net::TcpListener; +use std::process::ExitCode; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use serde_json::json; + +const USAGE: &str = "\ +usage: svc-gw [options] + --listen ADDR WebSocket listen address (0.0.0.0:7700) + --mock synthesise vehicle state, flagged as mock in every message + --recorder-ctl PATH recorder control socket, for 'mark' (/run/yukart-recorder/ctl.sock)"; + +pub fn now_us() -> i64 { + let d = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap_or_default(); + d.as_micros() as i64 +} + +pub fn mono_ns() -> i64 { + let mut ts: libc::timespec = unsafe { std::mem::zeroed() }; + unsafe { libc::clock_gettime(libc::CLOCK_MONOTONIC, &mut ts) }; + ts.tv_sec as i64 * 1_000_000_000 + ts.tv_nsec as i64 +} + +/// The car unit has an LTE uplink. The boundary serves the tablet (USB, 192.168.55.0/24), +/// the LAN and the tailnet — never whatever reaches the cellular interface. +fn allowed(ip: &std::net::IpAddr) -> bool { + match ip { + std::net::IpAddr::V4(v4) => { + let o = v4.octets(); + v4.is_loopback() || v4.is_private() || (o[0] == 100 && (64..128).contains(&o[1])) + } + std::net::IpAddr::V6(v6) => v6.is_loopback() || v6.to_ipv4_mapped().is_some_and(|v4| allowed(&v4.into())), + } +} + +fn main() -> ExitCode { + let mut listen = "0.0.0.0:7700".to_string(); + let mut mock = false; + let mut recorder_ctl = "/run/yukart-recorder/ctl.sock".to_string(); + let mut args = std::env::args().skip(1); + while let Some(a) = args.next() { + match a.as_str() { + "--mock" => mock = true, + "--listen" => listen = args.next().unwrap_or_default(), + "--recorder-ctl" => recorder_ctl = args.next().unwrap_or_default(), + _ => { + eprintln!("svc-gw: unknown option {a}\n{USAGE}"); + return ExitCode::from(2); + } + } + } + + let registry = signals::Registry::load(); + let hub = hub::Hub::default(); + let ctx = Arc::new(client::Ctx { + hub: hub.clone(), + hello: json!({ + "t": "hello", + "proto": 0, + "server": concat!("svc-gw ", env!("CARGO_PKG_VERSION")), + "host": std::fs::read_to_string("/proc/sys/kernel/hostname").unwrap_or_default().trim(), + "source": if mock { "mock" } else { "none" }, + "mock": mock, + "hb_ms": client::HEARTBEAT.as_millis() as u64, + "timeout_ms": client::TIMEOUT.as_millis() as u64, + }), + recorder_ctl, + acks: Mutex::new(Default::default()), + }); + + let listener = match TcpListener::bind(&listen) { + Ok(l) => l, + Err(e) => { + eprintln!("svc-gw: {listen}: {e}"); + return ExitCode::from(1); + } + }; + eprintln!("svc-gw {} on ws://{listen}{}", env!("CARGO_PKG_VERSION"), if mock { " — MOCK DATA" } else { "" }); + + // Steady state is reached with zero clients: the producer runs regardless. + thread::spawn(move || { + let mut m = mock::Mock::new(); + let mut seq: u64 = 0; + let mut last = Instant::now(); + let mut last_hb = Instant::now() - client::HEARTBEAT; + loop { + thread::sleep(Duration::from_millis(200)); + let dt = last.elapsed().as_secs_f64(); + last = Instant::now(); + if mock { + seq += 1; + let state = json!({ "t": "state", "seq": seq, "mock": true, "signals": m.step(®istry, dt) }); + hub.broadcast(&state.to_string()); + } + if last_hb.elapsed() >= client::HEARTBEAT { + last_hb = Instant::now(); + hub.broadcast(&json!({ "t": "hb", "ts": now_us() / 1000, "clients": hub.count() }).to_string()); + } + } + }); + + for conn in listener.incoming() { + let Ok(stream) = conn else { continue }; + match stream.peer_addr() { + Ok(a) if allowed(&a.ip()) => {} + Ok(a) => { + eprintln!("{a}: refused: not a local or tailnet address"); + continue; + } + Err(_) => continue, + } + if ctx.hub.count() >= 8 { + continue; // dropped: a display client limit, not a load-bearing one + } + let ctx = ctx.clone(); + thread::spawn(move || client::serve(stream, ctx)); + } + ExitCode::SUCCESS } diff --git a/src/mock.rs b/src/mock.rs new file mode 100644 index 0000000..a78d326 --- /dev/null +++ b/src/mock.rs @@ -0,0 +1,59 @@ +//! Synthesised vehicle state, for developing displays with no car. +//! +//! It degrades on purpose — signals go missing, stale and null — because that is +//! the normal state in a car, and a mock that is always complete and fresh trains +//! the display to assume something false. Every envelope carries `mock: true`. + +use serde_json::{json, Map, Value}; + +use crate::signals::Registry; + +pub struct Mock { + t: f64, + coolant: f64, +} + +impl Mock { + pub fn new() -> Self { + Mock { t: 0.0, coolant: 22.0 } + } + + /// One snapshot; `dt` seconds since the previous one. + pub fn step(&mut self, reg: &Registry, dt: f64) -> Map { + self.t += dt; + let t = self.t; + // A 60 s loop: idle, pull away, cruise, brake, idle. + let phase = t % 60.0; + let speed = match phase { + p if p < 10.0 => 0.0, + p if p < 25.0 => (p - 10.0) * 4.0, + p if p < 45.0 => 60.0 + 3.0 * (t * 0.7).sin(), + p if p < 55.0 => 60.0 * (55.0 - p) / 10.0, + _ => 0.0, + }; + let throttle = if (10.0..25.0).contains(&phase) { 35.0 } else if (25.0..45.0).contains(&phase) { 15.0 } else { 0.0 }; + let rpm = if speed < 1.0 { 820.0 + 25.0 * (t * 3.1).sin() } else { 1200.0 + speed * 35.0 }; + self.coolant = (self.coolant + dt * 0.4).min(86.0); + + let mut s = Map::new(); + let mut put = |name: &str, v: Value, q: Option<&str>| { + s.insert(name.to_string(), reg.envelope(name, v, q, true)); + }; + put("engine_rpm", json!(rpm.round()), None); + put("vehicle_speed_kph", json!((speed * 10.0_f64).round() / 10.0), None); + put("coolant_temp_c", json!(self.coolant.round()), None); + // Throttle drops out for 5 s in every 40: a signal that disappears. + if t % 40.0 >= 5.0 { + put("throttle_pct", json!(throttle), None); + } + // Intake temperature is stuck at its last value and marked stale for 10 s in every 30. + if t % 30.0 < 10.0 { + put("intake_air_temp_c", json!(31.0), Some("stale")); + } else { + put("intake_air_temp_c", json!((31.0 + (t * 0.05).sin()).round()), None); + } + // Fuel level is known but has no value: the registry has it, the car never sent it. + put("fuel_level_pct", Value::Null, None); + s + } +} diff --git a/src/signals.rs b/src/signals.rs new file mode 100644 index 0000000..4b9c998 --- /dev/null +++ b/src/signals.rs @@ -0,0 +1,45 @@ +//! Signal identity comes from contracts' registry, never authored here: the unit +//! in every envelope is looked up, so it cannot drift from the registry. +//! +//! `contracts` is read from a sibling checkout at build time. + +use serde_json::{json, Value}; +use std::collections::HashMap; + +const REGISTRY: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../contracts/schema/signals.json")); + +pub struct Registry { + units: HashMap, +} + +impl Registry { + pub fn load() -> Self { + let v: Value = serde_json::from_str(REGISTRY).expect("contracts signals.json is valid JSON"); + let units = v["signals"] + .as_object() + .expect("signals.json has a signals object") + .iter() + .map(|(k, s)| (k.clone(), s["unit"].as_str().unwrap_or("").to_string())) + .collect(); + Registry { units } + } + + /// An envelope (contracts `envelope.json`). Panics on a name the registry lacks: + /// that is a programming error, caught the first time the signal is produced. + pub fn envelope(&self, name: &str, value: Value, quality: Option<&str>, mock: bool) -> Value { + let unit = self.units.get(name).unwrap_or_else(|| panic!("signal {name} is not in contracts signals.json")); + let mut e = json!({ + "timestamp": crate::now_us(), + "monotonic": crate::mono_ns(), + "value": value, + "unit": unit, + }); + if let Some(q) = quality { + e["quality"] = json!(q); + } + if mock { + e["mock"] = json!(true); + } + e + } +}