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 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01DDwfd6Ki8UfG4hK6W5YpJz

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<VecDeque<(String, Value)>>,
+}
+
+pub fn serve(stream: TcpStream, ctx: Arc<Ctx>) {
+    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<Value> {
+    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<String> {
+    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())
+}

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<VecDeque<String>>,
+    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<String> {
+        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<Mutex<Vec<Arc<Outbox>>>>,
+}
+
+impl Hub {
+    pub fn join(&self) -> Arc<Outbox> {
+        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<Outbox>) {
+        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()
+    }
+}

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(&registry, 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
 }

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<String, Value> {
+        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
+    }
+}

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<String, String>,
+}
+
+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
+    }
+}