diff --git a/Cargo.lock b/Cargo.lock index a4d652e..e7df5c6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -563,6 +563,7 @@ dependencies = [ "rust-uci", "rustix", "serde", + "serde_json", "thiserror", "uci-derive", ] diff --git a/Cargo.toml b/Cargo.toml index 1ed5008..2b6ac00 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -29,6 +29,7 @@ flume = { version = "0.12.0", default-features = false, features = ["eventual-f config = { version = "0.15.25", default-features = false } csv = "1.4.0" serde = { version = "1", features = ["derive"] } +serde_json = { version = "1", optional = true } uci-derive = { path = "crates/uci-derive" } [dependencies.log] @@ -47,6 +48,7 @@ default = ["toml", "uci-vendored"] toml = ["config/toml"] uci = ["dep:rust-uci"] uci-vendored = ["uci", "rust-uci/vendored"] +trace = ["dep:serde_json"] [workspace] members = ["crates/uci-derive"] diff --git a/README.md b/README.md index 725d217..2d0fb05 100644 --- a/README.md +++ b/README.md @@ -10,6 +10,14 @@ * **Configurable via environment or UCI**: reads settings from environment variables or OpenWRT UCI, with sensible defaults and type‑safe parsing . * **Minimal dependencies**: relies on Rust crates only (anyhow, socket2, neli, rustix), no external scripts . +## Legacy control trace v1 + +The additive legacy characterization recorder is excluded from default builds. Build with +`--features trace` and set `SQMA_TRACE_V1_PATH` to a JSON Lines output path to enable it at +runtime. The format is versioned as trace v1 and is intended only for replaying the current +baseliner and rate-controller boundary during the `SqmControlLaw` extraction; it is not the +final runtime event schema. Existing CSV and metrics outputs remain unchanged. + ## Installation You can install **sqm‑autorate‑rust** in two ways: diff --git a/src/baseliner.rs b/src/baseliner.rs index 1f05116..319062b 100644 --- a/src/baseliner.rs +++ b/src/baseliner.rs @@ -9,6 +9,7 @@ use crate::SHUTDOWN; use crate::metrics::{Metric, MetricsSender}; use crate::pinger::PingReply; use crate::settings::Settings; +use crate::trace::{Recorder, v1}; use flume::{Receiver, Sender}; use log::{debug, info}; use std::collections::HashMap; @@ -78,6 +79,7 @@ pub struct Baseliner { pub selection_tx: Option>, pub baseline_metrics: MetricsSender, pub event_metrics: MetricsSender, + pub trace: Recorder, } fn ewma_factor(tick: f64, dur: f64) -> f64 { @@ -207,10 +209,23 @@ impl Baseliner { ); let snapshot = control_snapshot(&reflectors, Instant::now()); - if let Some(selection_tx) = &self.selection_tx { - let _ = selection_tx.send(snapshot.clone()); - } - self.control_tx.send(snapshot)?; + self.trace.linearize(|trace| { + if trace.is_enabled() { + trace.record_at( + reply.last_receive_time_s, + v1::Event::PingReply { + reflector: reply.reflector, + down_time_ms: reply.down_time, + up_time_ms: reply.up_time, + }, + ); + } + + if let Some(selection_tx) = &self.selection_tx { + let _ = selection_tx.send(snapshot.clone()); + } + self.control_tx.send(snapshot) + })?; } } } diff --git a/src/main.rs b/src/main.rs index 72392bf..a027eee 100644 --- a/src/main.rs +++ b/src/main.rs @@ -18,6 +18,7 @@ mod ratecontroller; mod reflector_selector; mod settings; mod time; +mod trace; mod util; use crate::baseliner::Baseliner; @@ -243,6 +244,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { } let start_time = Instant::now(); + let trace = trace::Recorder::from_env(start_time, settings); let probe_id = probe_identifier(); let ReflectorSetup { @@ -334,6 +336,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { selection_tx: selection_snapshot_tx, baseline_metrics, event_metrics: event_metrics.clone(), + trace: trace.clone(), }; let mut main_traffic_control = traffic_control_backend(); @@ -417,6 +420,7 @@ fn run(settings: &Settings) -> anyhow::Result<()> { reflector_peers, reselect_tx, rate_metrics, + trace.clone(), )?; let err_tx = error_tx.clone(); @@ -460,6 +464,8 @@ fn run(settings: &Settings) -> anyhow::Result<()> { &up_shaper, ); + trace.finish(); + result } diff --git a/src/ratecontroller.rs b/src/ratecontroller.rs index fb9798a..e988792 100644 --- a/src/ratecontroller.rs +++ b/src/ratecontroller.rs @@ -14,6 +14,7 @@ use crate::platform::{ }; use crate::settings::Settings; use crate::time::Time; +use crate::trace::{Recorder, v1}; use crate::util::{ArcRwLock, RwLockExt}; use flume::{Receiver, Sender}; use log::{debug, info, warn}; @@ -145,9 +146,11 @@ pub struct Ratecontroller { state_dl: State, state_ul: State, rng: fastrand::Rng, + next_evaluation_id: u64, stats_provider: S, traffic_control: T, metrics: MetricsSender, + trace: Recorder, } impl Ratecontroller { @@ -160,6 +163,7 @@ impl Ratecontroller { mut stats_provider: S, mut traffic_control: T, mut rng: fastrand::Rng, + trace: Recorder, ) -> anyhow::Result { let dl_shaper = traffic_control.find_shaper(settings.network.download_interface.as_str())?; @@ -178,7 +182,7 @@ impl Ratecontroller { let (cur_rx, cur_tx) = get_interface_stats(&mut stats_provider, &settings.network.upload_interface)?; - Ok(Self { + let controller = Self { settings, control_snapshot: None, control_snapshot_rx, @@ -187,10 +191,32 @@ impl Ratecontroller { state_dl: State::new(dl_shaper, cur_rx, dl_safe_rates), state_ul: State::new(ul_shaper, cur_tx, ul_safe_rates), rng, + next_evaluation_id: 0, stats_provider, traffic_control, metrics, - }) + trace, + }; + + if controller.trace.is_enabled() { + controller.trace.record_at( + Instant::now(), + v1::Event::ControllerInitialized { + previous_rx_bytes: controller.state_dl.previous_bytes, + previous_tx_bytes: controller.state_ul.previous_bytes, + download_prev_mono_ns: controller + .trace + .monotonic_ns(controller.state_dl.prev_t), + upload_prev_mono_ns: controller.trace.monotonic_ns(controller.state_ul.prev_t), + download_history_index: controller.state_dl.nrate as u64, + upload_history_index: controller.state_ul.nrate as u64, + download_safe_rates_kbit: controller.state_dl.safe_rates.clone(), + upload_safe_rates_kbit: controller.state_ul.safe_rates.clone(), + }, + ); + } + + Ok(controller) } fn request_initial_rates(&mut self) -> anyhow::Result<()> { @@ -198,6 +224,20 @@ impl Ratecontroller { self.state_dl.current_rate = self.settings.network.download_base_kbits * 0.6; self.state_ul.current_rate = self.settings.network.upload_base_kbits * 0.6; + if self.trace.is_enabled() { + self.trace.record_at( + Instant::now(), + v1::Event::RequestedRates { + evaluation_id: None, + reason: v1::RateRequestReason::Startup, + download_kbit: self.state_dl.current_rate.round() as u64, + upload_kbit: self.state_ul.current_rate.round() as u64, + download_requested: true, + upload_requested: true, + }, + ); + } + self.traffic_control.set_rate( &self.state_dl.shaper, self.state_dl.current_rate.round() as u64, @@ -212,7 +252,8 @@ impl Ratecontroller { Ok(()) } - fn calculate_rate(&mut self, direction: Direction) -> anyhow::Result<()> { + fn calculate_rate(&mut self, direction: Direction, evaluation_id: u64) -> anyhow::Result<()> { + let trace = self.trace.clone(); let (base_rate, delay_ms, min_rate, state) = if direction == Direction::Down { ( self.settings.network.download_base_kbits, @@ -231,6 +272,20 @@ impl Ratecontroller { let now_t = Instant::now(); let dur = now_t.duration_since(state.prev_t); + let trace_enabled = trace.is_enabled(); + let trace_direction = match direction { + Direction::Down => v1::Direction::Download, + Direction::Up => v1::Direction::Upload, + }; + if trace_enabled { + trace.record_at( + now_t, + v1::Event::RateCalculation { + evaluation_id, + direction: trace_direction, + }, + ); + } if !state.deltas.is_empty() { state.next_rate = state.current_rate; @@ -269,11 +324,20 @@ impl Ratecontroller { } if state.delta_stat > delay_ms { - match state - .safe_rates - .get(self.rng.usize(..state.safe_rates.len())) - { + let selected_index = self.rng.usize(..state.safe_rates.len()); + match state.safe_rates.get(selected_index) { Some(rnd_rate) => { + if trace_enabled { + trace.record_at( + Instant::now(), + v1::Event::RandomSafeRateChoice { + evaluation_id, + direction: trace_direction, + index: selected_index as u64, + rate_kbit: *rnd_rate, + }, + ); + } state.next_rate = rnd_rate.min(0.9 * state.current_rate * state.load); } None => { @@ -349,7 +413,9 @@ impl Ratecontroller { &mut self.stats_provider, &self.settings.network.upload_interface, )?; - self.update_deltas()?; + let evaluation_id = self.next_evaluation_id; + self.next_evaluation_id = self.next_evaluation_id.wrapping_add(1); + self.update_deltas(evaluation_id, now_t)?; if self.state_dl.deltas.is_empty() || self.state_ul.deltas.is_empty() { warn!("No reflector data available, dropping to minimum rates"); @@ -362,6 +428,20 @@ impl Ratecontroller { self.state_dl.next_rate = self.settings.network.download_min_kbits(); self.state_ul.next_rate = self.settings.network.upload_min_kbits(); + if self.trace.is_enabled() { + self.trace.record_at( + Instant::now(), + v1::Event::RequestedRates { + evaluation_id: Some(evaluation_id), + reason: v1::RateRequestReason::NoReflectorData, + download_kbit: self.state_dl.next_rate as u64, + upload_kbit: self.state_ul.next_rate as u64, + download_requested: true, + upload_requested: true, + }, + ); + } + self.traffic_control.set_rate( &self.state_dl.shaper, self.state_dl.next_rate as u64, @@ -378,8 +458,24 @@ impl Ratecontroller { continue; } - self.calculate_rate(Direction::Down)?; - self.calculate_rate(Direction::Up)?; + self.calculate_rate(Direction::Down, evaluation_id)?; + self.calculate_rate(Direction::Up, evaluation_id)?; + + if self.trace.is_enabled() { + let download_requested = self.state_dl.next_rate != self.state_dl.current_rate; + let upload_requested = self.state_ul.next_rate != self.state_ul.current_rate; + self.trace.record_at( + Instant::now(), + v1::Event::RequestedRates { + evaluation_id: Some(evaluation_id), + reason: v1::RateRequestReason::Control, + download_kbit: self.state_dl.next_rate as u64, + upload_kbit: self.state_ul.next_rate as u64, + download_requested, + upload_requested, + }, + ); + } if self.state_dl.next_rate != self.state_dl.current_rate || self.state_ul.next_rate != self.state_ul.current_rate @@ -479,18 +575,36 @@ impl Ratecontroller { } } - fn update_deltas(&mut self) -> anyhow::Result<()> { - drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot); - let now_t = Instant::now(); - let reflectors = self.reflectors_lock.read_anyhow()?; - let (down_deltas, up_deltas) = deltas_from_snapshot( - self.control_snapshot.as_ref(), - &reflectors, - now_t, - self.settings.advanced_settings.tick_interval, - ); - self.state_dl.deltas = down_deltas; - self.state_ul.deltas = up_deltas; + fn update_deltas(&mut self, evaluation_id: u64, loop_time: Instant) -> anyhow::Result<()> { + let recorder = self.trace.clone(); + recorder.linearize(|trace| -> anyhow::Result<()> { + drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot); + let now_t = Instant::now(); + let reflectors = self.reflectors_lock.read_anyhow()?; + let (down_deltas, up_deltas) = deltas_from_snapshot( + self.control_snapshot.as_ref(), + &reflectors, + now_t, + self.settings.advanced_settings.tick_interval, + ); + self.state_dl.deltas = down_deltas; + self.state_ul.deltas = up_deltas; + + if trace.is_enabled() { + trace.record_at( + now_t, + v1::Event::ControlEvaluation { + evaluation_id, + loop_mono_ns: self.trace.monotonic_ns(loop_time), + rx_bytes: self.state_dl.current_bytes, + tx_bytes: self.state_ul.current_bytes, + active_reflectors: reflectors.clone(), + }, + ); + } + + Ok(()) + })?; if self.state_dl.deltas.len() < 5 || self.state_ul.deltas.len() < 5 { // trigger reselection @@ -513,6 +627,7 @@ impl Ratecontroller { reflectors_lock: ArcRwLock>, reselect_trigger: Sender, metrics: MetricsSender, + trace: Recorder, ) -> anyhow::Result { Self::new_with_backends( settings, @@ -523,6 +638,7 @@ impl Ratecontroller { interface_stats_provider(), traffic_control_backend(), fastrand::Rng::new(), + trace, ) } } diff --git a/src/trace/mod.rs b/src/trace/mod.rs new file mode 100644 index 0000000..f626182 --- /dev/null +++ b/src/trace/mod.rs @@ -0,0 +1,417 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 + +pub(crate) mod v1; + +#[cfg(feature = "trace")] +mod writer; + +use crate::settings::Settings; +use std::time::Instant; + +#[cfg(feature = "trace")] +use std::time::SystemTime; + +#[cfg(feature = "trace")] +use flume::Sender; +#[cfg(feature = "trace")] +use log::warn; +#[cfg(feature = "trace")] +use std::fs::File; +#[cfg(feature = "trace")] +use std::io::{BufWriter, Write}; +#[cfg(feature = "trace")] +use std::path::Path; +#[cfg(feature = "trace")] +use std::sync::atomic::{AtomicBool, Ordering}; +#[cfg(feature = "trace")] +use std::sync::{Arc, Mutex}; +#[cfg(feature = "trace")] +use std::thread::JoinHandle; + +const TRACE_PATH_ENV: &str = "SQMA_TRACE_V1_PATH"; +#[cfg(feature = "trace")] +const QUEUE_CAPACITY: usize = 4_096; + +#[derive(Clone, Default)] +pub(crate) struct Recorder { + #[cfg(feature = "trace")] + inner: Option>, +} + +#[cfg(feature = "trace")] +struct Inner { + origin: Instant, + records: Sender, + next_seq: Mutex, + healthy: Arc, + writer: Mutex>>, +} + +pub(crate) struct OrderedRecorder<'a> { + #[cfg(feature = "trace")] + inner: Option<&'a Inner>, + #[cfg(feature = "trace")] + next_seq: Option<&'a mut u64>, + #[cfg(not(feature = "trace"))] + marker: std::marker::PhantomData<&'a ()>, +} + +impl Recorder { + pub(crate) fn from_env(origin: Instant, settings: &Settings) -> Self { + let Some(path) = std::env::var_os(TRACE_PATH_ENV) else { + return Self::disabled(); + }; + + #[cfg(feature = "trace")] + { + let header = v1::Event::Header { + format: v1::FORMAT_NAME.to_owned(), + version: v1::FORMAT_VERSION, + program_version: env!("CARGO_PKG_VERSION").to_owned(), + control: v1::ControlConfig::from_settings(settings), + }; + Self::start_path(Path::new(&path), origin, header) + } + + #[cfg(not(feature = "trace"))] + { + let _ = (origin, settings, path); + log::warn!( + "{TRACE_PATH_ENV} is set, but this binary was built without the `trace` feature" + ); + Self::disabled() + } + } + + pub(crate) fn disabled() -> Self { + Self::default() + } + + pub(crate) fn is_enabled(&self) -> bool { + #[cfg(feature = "trace")] + { + self.inner + .as_ref() + .is_some_and(|inner| inner.healthy.load(Ordering::Acquire)) + } + + #[cfg(not(feature = "trace"))] + { + false + } + } + + pub(crate) fn monotonic_ns(&self, instant: Instant) -> u64 { + #[cfg(feature = "trace")] + { + self.inner + .as_ref() + .map_or(0, |inner| v1::monotonic_ns(inner.origin, instant)) + } + + #[cfg(not(feature = "trace"))] + { + let _ = instant; + 0 + } + } + + pub(crate) fn record_at(&self, instant: Instant, event: v1::Event) -> Option { + self.linearize(|recorder| recorder.record_at(instant, event)) + } + + pub(crate) fn linearize(&self, callback: impl FnOnce(&mut OrderedRecorder<'_>) -> R) -> R { + #[cfg(feature = "trace")] + { + if self + .inner + .as_ref() + .is_some_and(|inner| inner.healthy.load(Ordering::Acquire)) + { + let inner = self + .inner + .as_ref() + .expect("healthy trace recorder must have inner state"); + match inner.next_seq.lock() { + Ok(mut next_seq) => { + if inner.healthy.load(Ordering::Acquire) { + let mut recorder = OrderedRecorder { + inner: Some(inner), + next_seq: Some(&mut next_seq), + }; + return callback(&mut recorder); + } + } + Err(_) => { + if inner.healthy.swap(false, Ordering::AcqRel) { + warn!("Legacy control trace sequencing lock failed; tracing disabled"); + } + } + } + } + + let mut recorder = OrderedRecorder { + inner: None, + next_seq: None, + }; + callback(&mut recorder) + } + + #[cfg(not(feature = "trace"))] + { + let mut recorder = OrderedRecorder { + marker: std::marker::PhantomData, + }; + callback(&mut recorder) + } + } + + pub(crate) fn finish(&self) { + #[cfg(feature = "trace")] + if let Some(inner) = self.inner.as_ref() { + self.linearize(|recorder| { + recorder.record_at(Instant::now(), v1::Event::End { clean: true }); + inner.healthy.store(false, Ordering::Release); + }); + + let writer = match inner.writer.lock() { + Ok(mut writer) => writer.take(), + Err(_) => { + warn!("Legacy control trace writer lock failed during shutdown"); + None + } + }; + let writer_panicked = writer.map(|writer| writer.join().is_err()).unwrap_or(false); + if writer_panicked { + warn!("Legacy control trace writer panicked during shutdown"); + } + } + } + + #[cfg(feature = "trace")] + fn start_path(path: &Path, origin: Instant, header: v1::Event) -> Self { + let file = match File::options() + .create(true) + .write(true) + .truncate(true) + .open(path) + { + Ok(file) => file, + Err(error) => { + warn!( + "Failed to open legacy control trace at {}: {error}; tracing disabled", + path.display() + ); + return Self::disabled(); + } + }; + + Self::start_writer(BufWriter::new(file), origin, header) + } + + #[cfg(feature = "trace")] + fn start_writer(writer: W, origin: Instant, header: v1::Event) -> Self + where + W: Write + Send + 'static, + { + let (records, record_rx) = flume::bounded(QUEUE_CAPACITY); + let healthy = Arc::new(AtomicBool::new(true)); + let header = v1::Record { + seq: 0, + timestamp: v1::timestamp(origin, origin, SystemTime::now()), + event: header, + }; + let writer = match writer::spawn(writer, record_rx, header, Arc::clone(&healthy)) { + Ok(writer) => writer, + Err(error) => { + warn!("Failed to start legacy control trace writer: {error}; tracing disabled"); + return Self::disabled(); + } + }; + + Self { + inner: Some(Arc::new(Inner { + origin, + records, + next_seq: Mutex::new(1), + healthy, + writer: Mutex::new(Some(writer)), + })), + } + } +} + +impl OrderedRecorder<'_> { + pub(crate) fn is_enabled(&self) -> bool { + #[cfg(feature = "trace")] + { + self.inner.is_some() + } + + #[cfg(not(feature = "trace"))] + { + false + } + } + + pub(crate) fn record_at(&mut self, instant: Instant, event: v1::Event) -> Option { + #[cfg(feature = "trace")] + { + let inner = self.inner?; + let next_seq = self.next_seq.as_deref_mut()?; + let seq = *next_seq; + let record = v1::Record { + seq, + timestamp: v1::timestamp(inner.origin, instant, SystemTime::now()), + event, + }; + + if inner.records.send(record).is_err() { + if inner.healthy.swap(false, Ordering::AcqRel) { + warn!("Legacy control trace channel closed; tracing disabled"); + } + return None; + } + + *next_seq = (*next_seq).saturating_add(1); + Some(seq) + } + + #[cfg(not(feature = "trace"))] + { + let _ = (instant, event); + None + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn disabled_recorder_is_a_no_op() { + let recorder = Recorder::disabled(); + + assert!(!recorder.is_enabled()); + assert_eq!( + recorder.record_at(Instant::now(), v1::Event::End { clean: true }), + None + ); + recorder.finish(); + } + + #[cfg(feature = "trace")] + mod enabled { + use super::*; + use std::io; + use std::net::{IpAddr, Ipv4Addr}; + use std::path::PathBuf; + use std::sync::Barrier; + use std::sync::atomic::{AtomicU64, Ordering}; + use std::thread; + + static NEXT_PATH_ID: AtomicU64 = AtomicU64::new(0); + + struct FailingWriter; + + impl Write for FailingWriter { + fn write(&mut self, _buffer: &[u8]) -> io::Result { + Err(io::Error::other("injected writer failure")) + } + + fn flush(&mut self) -> io::Result<()> { + Err(io::Error::other("injected writer failure")) + } + } + + fn temp_path() -> PathBuf { + let id = NEXT_PATH_ID.fetch_add(1, Ordering::Relaxed); + std::env::temp_dir().join(format!( + "sqm-autorate-trace-v1-{}-{id}.jsonl", + std::process::id() + )) + } + + fn ping_event(reflector: Ipv4Addr, value: f64) -> v1::Event { + v1::Event::PingReply { + reflector: IpAddr::V4(reflector), + down_time_ms: value, + up_time_ms: value, + } + } + + #[test] + fn writer_preserves_global_sequence_order_from_multiple_producers() { + let path = temp_path(); + let origin = Instant::now(); + let recorder = Recorder::start_path(&path, origin, v1::tests::header_event()); + let barrier = Arc::new(Barrier::new(3)); + let mut producers = Vec::new(); + + for producer in 1..=2 { + let recorder = recorder.clone(); + let barrier = Arc::clone(&barrier); + producers.push(thread::spawn(move || { + barrier.wait(); + for value in 0..50 { + recorder.record_at( + origin, + ping_event(Ipv4Addr::new(192, 0, 2, producer), value as f64), + ); + } + })); + } + + barrier.wait(); + for producer in producers { + producer.join().unwrap(); + } + recorder.finish(); + + let contents = std::fs::read_to_string(&path).unwrap(); + let records = contents + .lines() + .map(|line| serde_json::from_str::(line).unwrap()) + .collect::>(); + + assert_eq!(records.len(), 102); + assert!(matches!( + &records.first().unwrap().event, + v1::Event::Header { .. } + )); + assert!(matches!( + &records.last().unwrap().event, + v1::Event::End { clean: true } + )); + for (expected, record) in records.iter().enumerate() { + assert_eq!(record.seq, expected as u64); + } + + std::fs::remove_file(path).unwrap(); + } + + #[test] + fn writer_failure_disables_recorder_without_propagating() { + let recorder = + Recorder::start_writer(FailingWriter, Instant::now(), v1::tests::header_event()); + + recorder.record_at(Instant::now(), ping_event(Ipv4Addr::new(192, 0, 2, 1), 1.0)); + recorder.finish(); + + assert!(!recorder.is_enabled()); + } + + #[test] + fn open_failure_returns_disabled_recorder() { + let path = temp_path().join("missing-parent").join("trace.jsonl"); + let recorder = Recorder::start_path(&path, Instant::now(), v1::tests::header_event()); + + assert!(!recorder.is_enabled()); + recorder.record_at(Instant::now(), v1::Event::End { clean: true }); + recorder.finish(); + } + } +} diff --git a/src/trace/v1.rs b/src/trace/v1.rs new file mode 100644 index 0000000..3c7f94a --- /dev/null +++ b/src/trace/v1.rs @@ -0,0 +1,234 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 + +use serde::{Deserialize, Serialize}; +use std::net::IpAddr; + +#[cfg(feature = "trace")] +use crate::settings::Settings; +#[cfg(any(feature = "trace", test))] +use std::time::Duration; +#[cfg(any(feature = "trace", test))] +use std::time::Instant; +#[cfg(feature = "trace")] +use std::time::{SystemTime, UNIX_EPOCH}; + +#[cfg(feature = "trace")] +pub(crate) const FORMAT_VERSION: u16 = 1; +#[cfg(feature = "trace")] +pub(crate) const FORMAT_NAME: &str = "legacy-control-trace-v1"; + +#[cfg(feature = "trace")] +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub(crate) struct Timestamp { + pub(crate) mono_ns: u64, + pub(crate) unix_us: i64, +} + +#[cfg(feature = "trace")] +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub(crate) struct Record { + pub(crate) seq: u64, + pub(crate) timestamp: Timestamp, + #[serde(flatten)] + pub(crate) event: Event, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +#[serde(tag = "type", rename_all = "snake_case")] +pub(crate) enum Event { + #[cfg(feature = "trace")] + Header { + format: String, + version: u16, + program_version: String, + control: ControlConfig, + }, + PingReply { + reflector: IpAddr, + down_time_ms: f64, + up_time_ms: f64, + }, + ControllerInitialized { + previous_rx_bytes: i128, + previous_tx_bytes: i128, + download_prev_mono_ns: u64, + upload_prev_mono_ns: u64, + download_history_index: u64, + upload_history_index: u64, + download_safe_rates_kbit: Vec, + upload_safe_rates_kbit: Vec, + }, + ControlEvaluation { + evaluation_id: u64, + loop_mono_ns: u64, + rx_bytes: i128, + tx_bytes: i128, + active_reflectors: Vec, + }, + RateCalculation { + evaluation_id: u64, + direction: Direction, + }, + RandomSafeRateChoice { + evaluation_id: u64, + direction: Direction, + index: u64, + rate_kbit: f64, + }, + RequestedRates { + evaluation_id: Option, + reason: RateRequestReason, + download_kbit: u64, + upload_kbit: u64, + download_requested: bool, + upload_requested: bool, + }, + End { + clean: bool, + }, +} + +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum Direction { + Download, + Upload, +} + +#[derive(Clone, Copy, Debug, Deserialize, PartialEq, Serialize)] +#[serde(rename_all = "snake_case")] +pub(crate) enum RateRequestReason { + Startup, + Control, + NoReflectorData, +} + +#[cfg(feature = "trace")] +#[derive(Clone, Debug, Deserialize, PartialEq, Serialize)] +pub(crate) struct ControlConfig { + pub(crate) tick_interval_secs: f64, + pub(crate) min_change_interval_secs: f64, + pub(crate) download_base_kbit: f64, + pub(crate) upload_base_kbit: f64, + pub(crate) download_min_kbit: f64, + pub(crate) upload_min_kbit: f64, + pub(crate) download_delay_ms: f64, + pub(crate) upload_delay_ms: f64, + pub(crate) high_load_level: f64, + pub(crate) speed_history_size: u32, +} + +#[cfg(feature = "trace")] +impl ControlConfig { + pub(crate) fn from_settings(settings: &Settings) -> Self { + Self { + tick_interval_secs: settings.advanced_settings.tick_interval, + min_change_interval_secs: settings.advanced_settings.min_change_interval, + download_base_kbit: settings.network.download_base_kbits, + upload_base_kbit: settings.network.upload_base_kbits, + download_min_kbit: settings.network.download_min_kbits(), + upload_min_kbit: settings.network.upload_min_kbits(), + download_delay_ms: settings.advanced_settings.download_delay_ms, + upload_delay_ms: settings.advanced_settings.upload_delay_ms, + high_load_level: settings.advanced_settings.high_load_level, + speed_history_size: settings.advanced_settings.speed_hist_size, + } + } +} + +#[cfg(feature = "trace")] +pub(crate) fn timestamp(origin: Instant, monotonic: Instant, realtime: SystemTime) -> Timestamp { + Timestamp { + mono_ns: monotonic_ns(origin, monotonic), + unix_us: unix_us(realtime), + } +} + +#[cfg(any(feature = "trace", test))] +pub(crate) fn monotonic_ns(origin: Instant, monotonic: Instant) -> u64 { + duration_ns(monotonic.saturating_duration_since(origin)) +} + +#[cfg(any(feature = "trace", test))] +fn duration_ns(duration: Duration) -> u64 { + duration.as_nanos().min(u64::MAX as u128) as u64 +} + +#[cfg(feature = "trace")] +fn unix_us(realtime: SystemTime) -> i64 { + match realtime.duration_since(UNIX_EPOCH) { + Ok(duration) => duration + .as_micros() + .min(i64::MAX as u128) + .try_into() + .unwrap_or(i64::MAX), + Err(error) => { + let micros = error.duration().as_micros().min(i64::MAX as u128) as i64; + -micros + } + } +} + +#[cfg(test)] +pub(super) mod tests { + use super::*; + + #[cfg(feature = "trace")] + fn control_config() -> ControlConfig { + ControlConfig { + tick_interval_secs: 0.5, + min_change_interval_secs: 0.5, + download_base_kbit: 100_000.0, + upload_base_kbit: 20_000.0, + download_min_kbit: 20_000.0, + upload_min_kbit: 4_000.0, + download_delay_ms: 15.0, + upload_delay_ms: 15.0, + high_load_level: 0.8, + speed_history_size: 100, + } + } + + #[cfg(feature = "trace")] + pub(crate) fn header_event() -> Event { + Event::Header { + format: FORMAT_NAME.to_owned(), + version: FORMAT_VERSION, + program_version: "test".to_owned(), + control: control_config(), + } + } + + #[cfg(feature = "trace")] + #[test] + fn serializes_versioned_header() { + let origin = Instant::now(); + let record = Record { + seq: 0, + timestamp: timestamp(origin, origin, UNIX_EPOCH + Duration::from_secs(1)), + event: header_event(), + }; + + let json = serde_json::to_value(&record).unwrap(); + + assert_eq!(json["seq"], 0); + assert_eq!(json["type"], "header"); + assert_eq!(json["format"], FORMAT_NAME); + assert_eq!(json["version"], FORMAT_VERSION); + assert_eq!(json["timestamp"]["mono_ns"], 0); + assert_eq!(serde_json::from_value::(json).unwrap(), record); + } + + #[test] + fn converts_existing_instants_relative_to_origin() { + let origin = Instant::now(); + + assert_eq!( + monotonic_ns(origin, origin + Duration::from_nanos(123_456)), + 123_456 + ); + assert_eq!(monotonic_ns(origin, origin - Duration::from_nanos(1)), 0); + } +} diff --git a/src/trace/writer.rs b/src/trace/writer.rs new file mode 100644 index 0000000..48ee252 --- /dev/null +++ b/src/trace/writer.rs @@ -0,0 +1,54 @@ +// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair) +// +// SPDX-License-Identifier: MPL-2.0 + +use super::v1::{Event, Record}; +use flume::Receiver; +use log::warn; +use std::io::{self, Write}; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::thread::{self, JoinHandle}; + +pub(super) fn spawn( + writer: W, + records: Receiver, + header: Record, + healthy: Arc, +) -> io::Result> +where + W: Write + Send + 'static, +{ + thread::Builder::new() + .name("trace-v1-writer".to_owned()) + .spawn(move || { + if let Err(error) = write_records(writer, records, header) { + warn!("Legacy control trace writer failed; tracing disabled: {error}"); + healthy.store(false, Ordering::Release); + } + }) +} + +fn write_records(mut writer: W, records: Receiver, header: Record) -> io::Result<()> +where + W: Write, +{ + write_record(&mut writer, &header)?; + + while let Ok(record) = records.recv() { + let clean_end = matches!(record.event, Event::End { clean: true }); + write_record(&mut writer, &record)?; + + if clean_end { + writer.flush()?; + return Ok(()); + } + } + + writer.flush() +} + +fn write_record(writer: &mut impl Write, record: &Record) -> io::Result<()> { + serde_json::to_writer(&mut *writer, record).map_err(io::Error::other)?; + writer.write_all(b"\n") +}