diff --git a/Cargo.lock b/Cargo.lock index 372bf8e..4922685 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -75,6 +75,54 @@ dependencies = [ "crypto-common", ] +[[package]] +name = "fixedbitset" +version = "0.5.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99" + +[[package]] +name = "flume" +version = "0.11.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "da0e4dd2a88388a1f4ccc7c9ce104604dab68d9f408dc34cd45823d5a9069095" +dependencies = [ + "futures-core", + "futures-sink", + "spin", +] + +[[package]] +name = "futures-core" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "92d699e522242e69e3003b94ecc1f960f3a5e015aa7c5d7486e65ad01dd94f5e" + +[[package]] +name = "futures-sink" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1944426bf7d03f1d14f708785e4b33efd750b36d48a157b836b3efc15ede8e1d" + +[[package]] +name = "futures-task" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd" + +[[package]] +name = "futures-util" +version = "0.3.34" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0d50a92467f8ba5dd6e3ee5d4bd04d73ab2e4e1c44474a0674821dfce14b79bc" +dependencies = [ + "futures-core", + "futures-sink", + "futures-task", + "pin-project-lite", + "slab", +] + [[package]] name = "getrandom" version = "0.4.3" @@ -124,6 +172,15 @@ version = "0.2.190" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" +[[package]] +name = "lock_api" +version = "0.4.14" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "224399e74b87b5f3557511d98dff8b14089b3dadafcab6bb93eab67d3aace965" +dependencies = [ + "scopeguard", +] + [[package]] name = "log" version = "0.4.34" @@ -136,6 +193,23 @@ version = "2.8.3" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cf8baf1c55e62ffcace7a9f06f4bd9cd3f0c4beb022d3b367256b91b87513d98" +[[package]] +name = "mio" +version = "1.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1788edb87fdc09c7e26304471e2f5be8cdefb1b6930d6e3985fc02ff53bf86ee" +dependencies = [ + "libc", + "wasi", + "windows-sys", +] + +[[package]] +name = "pin-project-lite" +version = "0.2.17" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" + [[package]] name = "proc-macro2" version = "1.0.107" @@ -177,6 +251,29 @@ version = "0.10.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" +[[package]] +name = "rumqttc" +version = "0.25.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0feff8d882bff0b2fddaf99355a10336d43dd3ed44204f85ece28cf9626ab519" +dependencies = [ + "bytes", + "fixedbitset", + "flume", + "futures-util", + "log", + "thiserror", + "tokio", + "tokio-stream", + "tokio-util", +] + +[[package]] +name = "scopeguard" +version = "1.2.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" + [[package]] name = "serde" version = "1.0.229" @@ -230,11 +327,37 @@ dependencies = [ "digest", ] +[[package]] +name = "slab" +version = "0.4.12" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0c790de23124f9ab44544d7ac05d60440adc586479ce501c1d6d7da3cd8c9cf5" + +[[package]] +name = "socket2" +version = "0.6.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c3d1e2c7f27f8d4cb10542a02c49005dbd6e93095799d6f3be745fae9f8fedd4" +dependencies = [ + "libc", + "windows-sys", +] + +[[package]] +name = "spin" +version = "0.9.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3763264f6b73151db08c50ff20d7d8a0b8796e021cdea7ceedad07b80155fa0e" +dependencies = [ + "lock_api", +] + [[package]] name = "svc-gw" version = "0.1.0" dependencies = [ "libc", + "rumqttc", "serde_json", "tungstenite", ] @@ -270,6 +393,57 @@ dependencies = [ "syn", ] +[[package]] +name = "tokio" +version = "1.53.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e95f91fcc7a621e8b030f6aa23c71fe9838ae2fb4d8118b75602a328f5144044" +dependencies = [ + "bytes", + "libc", + "mio", + "pin-project-lite", + "socket2", + "tokio-macros", + "windows-sys", +] + +[[package]] +name = "tokio-macros" +version = "2.7.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "78773a2a397f451582ce068015985c33193cf6dea8b74d2a639fe457b2f07b0e" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + +[[package]] +name = "tokio-stream" +version = "0.1.19" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a3d06f0b082ba57c26b79407372e57cf2a1e28124f78e9479fe80322cf53420b" +dependencies = [ + "futures-core", + "pin-project-lite", + "tokio", +] + +[[package]] +name = "tokio-util" +version = "0.7.20" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e464cf451ba96ebfc6f9b6542f17ee8b8956e33f1e40d9690624e59d7a7f8a4b" +dependencies = [ + "bytes", + "futures-core", + "futures-sink", + "libc", + "pin-project-lite", + "tokio", +] + [[package]] name = "tungstenite" version = "0.30.0" @@ -298,6 +472,27 @@ version = "1.0.26" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d245f478577f809a851594d02313b640fb437e0bb33866753cff937863096954" +[[package]] +name = "wasi" +version = "0.11.1+wasi-snapshot-preview1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b" + +[[package]] +name = "windows-link" +version = "0.2.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f0805222e57f7521d6a62e36fa9163bc891acd422f971defe97d64e70d0a4fe5" + +[[package]] +name = "windows-sys" +version = "0.61.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc" +dependencies = [ + "windows-link", +] + [[package]] name = "zmij" version = "1.0.23" diff --git a/Cargo.toml b/Cargo.toml index c7c6e16..21802f1 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -7,6 +7,7 @@ publish = false [dependencies] libc = "0.2" +rumqttc = { version = "0.25", default-features = false } serde_json = "1" tungstenite = { version = "0.30", default-features = false, features = ["handshake"] } diff --git a/deploy/yukart-gw.service b/deploy/yukart-gw.service index 72be14b..ab9c238 100644 --- a/deploy/yukart-gw.service +++ b/deploy/yukart-gw.service @@ -1,6 +1,8 @@ [Unit] Description=Yukart gateway: the display boundary (svc-gw) Documentation=gitea:yukart/svc-gw +# Tier 3. Ordered after the broker and nothing else; it reconnects by itself if the broker restarts. +After=mosquitto.service StartLimitIntervalSec=0 [Service] diff --git a/docs/impl/plans/INITIAL.md b/docs/impl/plans/INITIAL.md index 7e3defc..a1aa0ce 100644 --- a/docs/impl/plans/INITIAL.md +++ b/docs/impl/plans/INITIAL.md @@ -1,7 +1,7 @@ --- type: plan status: active -updated: 2026-10-09 +updated: 2026-10-12 --- # Initial plan @@ -23,6 +23,9 @@ directly. 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). + **Bus input** — in place (2026-10-12): subscribes to `yukart/vehicle/+` and + `yukart/host/+`, keeps the newest envelope per signal, marks stale after 5 s + or when known only from a retained copy. Bus or mock, never both. 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. diff --git a/src/bus.rs b/src/bus.rs new file mode 100644 index 0000000..d0a4831 --- /dev/null +++ b/src/bus.rs @@ -0,0 +1,88 @@ +//! The real input: envelopes from the bus, latest per signal. +//! +//! Subscribes to the envelope topics and keeps the newest envelope of each signal, +//! keyed by the topic's last segment (the signal name, contracts guideline 1). +//! Staleness is judged here, by arrival: a signal not refreshed within `STALE`, +//! or known only from the broker's retained copy (it may predate a reboot), is +//! sent with `quality: "stale"`. Signals are never dropped once seen; an old value +//! shown stale is the honest picture. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::thread; +use std::time::{Duration, Instant}; + +use rumqttc::{Client, Event, MqttOptions, Packet, QoS}; +use serde_json::{json, Map, Value}; + +const TOPICS: [&str; 2] = ["yukart/vehicle/+", "yukart/host/+"]; +const STALE: Duration = Duration::from_secs(5); + +struct Entry { + envelope: Value, + at: Instant, + retained: bool, +} + +#[derive(Clone)] +pub struct Bus { + latest: Arc>>, +} + +impl Bus { + /// Connects in the background and reconnects forever; never blocks the caller. + pub fn start(host: &str, port: u16) -> Self { + let bus = Bus { latest: Arc::default() }; + let mut opts = MqttOptions::new("svc-gw", host, port); + opts.set_keep_alive(Duration::from_secs(5)); + let (client, mut connection) = Client::new(opts, 64); + let latest = bus.latest.clone(); + thread::spawn(move || { + let mut up = false; + for ev in connection.iter() { + match ev { + Ok(Event::Incoming(Packet::ConnAck(_))) => { + up = true; + eprintln!("bus: connected; subscribing to {}", TOPICS.join(", ")); + // A clean session forgets subscriptions: renew them on every connect. + for t in TOPICS { + let _ = client.try_subscribe(t, QoS::AtMostOnce); + } + } + Ok(Event::Incoming(Packet::Publish(p))) => { + let Some(name) = p.topic.rsplit('/').next() else { continue }; + let Ok(envelope) = serde_json::from_slice::(&p.payload) else { continue }; + if !envelope.is_object() { + continue; + } + let entry = Entry { envelope, at: Instant::now(), retained: p.retain }; + latest.lock().unwrap().insert(name.to_string(), entry); + } + Ok(_) => {} + Err(e) => { + if std::mem::take(&mut up) { + eprintln!("bus: {e}; retrying"); + } + thread::sleep(Duration::from_secs(1)); + } + } + } + }); + bus + } + + /// Every signal seen, newest envelope each, marked stale where it is. + pub fn snapshot(&self) -> Map { + let latest = self.latest.lock().unwrap(); + latest + .iter() + .map(|(name, e)| { + let mut env = e.envelope.clone(); + if (e.retained || e.at.elapsed() > STALE) && env.get("quality").is_none() { + env["quality"] = json!("stale"); + } + (name.clone(), env) + }) + .collect() + } +} diff --git a/src/main.rs b/src/main.rs index 669f8ad..798ad63 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,8 +2,10 @@ //! 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. +//! State comes from the bus (`bus.rs`), or from the mock with `--mock`: one input +//! or the other, never both, so mock data cannot mix into real data. +mod bus; mod client; mod hub; mod mock; @@ -20,7 +22,8 @@ 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 + --mock synthesise vehicle state, flagged as mock in every message (no bus) + --broker HOST:PORT MQTT broker; '' for none (127.0.0.1:1883) --recorder-ctl PATH recorder control socket, for 'mark' (/run/yukart-recorder/ctl.sock)"; pub fn now_us() -> i64 { @@ -50,12 +53,14 @@ 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 broker = "127.0.0.1:1883".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(), + "--broker" => broker = args.next().unwrap_or_default(), _ => { eprintln!("svc-gw: unknown option {a}\n{USAGE}"); return ExitCode::from(2); @@ -64,6 +69,18 @@ fn main() -> ExitCode { } let registry = signals::Registry::load(); + let bus = if mock || broker.is_empty() { + None + } else { + match broker.rsplit_once(':').and_then(|(h, p)| Some((h, p.parse::().ok()?))) { + Some((host, port)) => Some(bus::Bus::start(host, port)), + None => { + eprintln!("svc-gw: --broker must be HOST:PORT\n{USAGE}"); + return ExitCode::from(2); + } + } + }; + let source = if mock { "mock" } else if bus.is_some() { "bus" } else { "none" }; let hub = hub::Hub::default(); let ctx = Arc::new(client::Ctx { hub: hub.clone(), @@ -72,7 +89,7 @@ fn main() -> ExitCode { "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" }, + "source": source, "mock": mock, "hb_ms": client::HEARTBEAT.as_millis() as u64, "timeout_ms": client::TIMEOUT.as_millis() as u64, @@ -88,7 +105,7 @@ fn main() -> ExitCode { return ExitCode::from(1); } }; - eprintln!("svc-gw {} on ws://{listen}{}", env!("CARGO_PKG_VERSION"), if mock { " — MOCK DATA" } else { "" }); + eprintln!("svc-gw {} on ws://{listen}, source {source}{}", env!("CARGO_PKG_VERSION"), if mock { " — MOCK DATA" } else { "" }); // Steady state is reached with zero clients: the producer runs regardless. thread::spawn(move || { @@ -100,9 +117,14 @@ fn main() -> ExitCode { thread::sleep(Duration::from_millis(200)); let dt = last.elapsed().as_secs_f64(); last = Instant::now(); - if mock { + let signals = match &bus { + Some(b) => Some(b.snapshot()).filter(|s| !s.is_empty()), + None if mock => Some(m.step(®istry, dt)), + None => None, + }; + if let Some(signals) = signals { seq += 1; - let state = json!({ "t": "state", "seq": seq, "mock": true, "signals": m.step(®istry, dt) }); + let state = json!({ "t": "state", "seq": seq, "mock": mock, "signals": signals }); hub.broadcast(&state.to_string()); } if last_hb.elapsed() >= client::HEARTBEAT {