chore(deps): update google/osv-scanner-action action to v2.5.1

Reviewed-on: #76

.github/workflows/build.yml

@@ -5,6 +5,7 @@ on:
     branches: [ "master" ]
   pull_request:
     branches: [ "master" ]
+  workflow_dispatch:
 
 env:
   CARGO_TERM_COLOR: always
@@ -18,5 +19,8 @@ jobs:
     - uses: actions/checkout@v7
     - name: Install Mold
       run: sudo apt install mold
+    - name: Install Rust toolchain
+      uses: actions-rust-lang/setup-rust-toolchain@166cdcfd11aee3cb47222f9ddb555ce30ddb9659 # v1
+    
     - name: Build
       run: cargo build --verbose

.github/workflows/osv-scanner.yml

@@ -34,7 +34,7 @@ jobs:
       actions: read
     if: ${{ github.event_name == 'push' || github.event_name == 'schedule' }}
     # If you want to copy this config, highly suggest pinning this version to a release rather than tracking the main branch
-    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@v2.5.0"
+    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable.yml@v2.5.1"
     with:
       # Just scan the root directory and docs, since everything else is fixtures
       scan-args: |-
@@ -47,7 +47,7 @@ jobs:
       actions: read
     if: ${{ github.event_name == 'pull_request' || github.event_name == 'merge_group' }}
     # If you want to copy this config, highly suggest pinning this version to a release rather than tracking the main branch
-    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.5.0"
+    uses: "google/osv-scanner-action/.github/workflows/osv-scanner-reusable-pr.yml@v2.5.1"
     with:
       # Just scan the root directory and docs, since everything else is fixtures
       scan-args: |-

Cargo.lock

@@ -315,15 +315,15 @@ checksum = "68354c5c6bd36d73ff3feceb05efa59b6acb7626617f4962be322a825e61f79a"
 
 [[package]]
 name = "netlink-bindings"
-version = "0.3.3"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "114816d03a633cef7cfa951721dafbda16bb1ff6af83393ee568ec2d4e4ded2d"
+checksum = "2a2a0dd8beef757878b977fd3d662c82a93dac4bf7632f42577ca8f8f4d701a0"
 
 [[package]]
 name = "netlink-socket2"
-version = "0.3.3"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "11193969f046efafdc9d8670d395a6ddcf9b806169f504e520a35dc95cab3241"
+checksum = "5c7ddb0ea7aa692947d5105160acb7c75c17bc5ec6d5a6263440ca7db4be9a34"
 dependencies = [
  "libc",
  "maybe-async",
@@ -563,7 +563,6 @@ dependencies = [
  "rust-uci",
  "rustix",
  "serde",
- "serde_json",
  "thiserror",
  "uci-derive",
 ]
@@ -607,18 +606,18 @@ dependencies = [
 
 [[package]]
 name = "thiserror"
-version = "2.0.19"
+version = "2.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09a43598840e33d5b0331f38c5e30d13bb11c11210a4b58f0d9b18a5a5eefcd9"
+checksum = "ec86235f5fcc2a73650310756d2ac5b138a5780bbbdfae3eeccec992c435ba4f"
 dependencies = [
  "thiserror-impl",
 ]
 
 [[package]]
 name = "thiserror-impl"
-version = "2.0.19"
+version = "2.0.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "43cbfe0cf76104d42a574802844187e84a305e531ed54455f11fbde0f10541cd"
+checksum = "bc04cd3e1236dd4a98afca4569f2deb3f120e5422a4023be2cb683f8486292af"
 dependencies = [
  "proc-macro2",
  "quote",

Cargo.toml

@@ -29,7 +29,6 @@ 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]
@@ -44,11 +43,10 @@ netlink-socket2 = { version = "0.3.3" }
 rust-uci = { git = "https://github.com/Lochnair/rust-uci.git", rev = "fcfedd204fb76928ef617bf9631b041c816e03d4", optional = true, default-features = false, }
 
 [features]
-default = ["toml", "trace", "uci-vendored"]
+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"]

README.md

@@ -10,14 +10,6 @@
 * **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:

renovate.json

@@ -2,5 +2,18 @@
   "$schema": "https://docs.renovatebot.com/renovate-schema.json",
   "extends": [
     "config:recommended"
+  ],
+  "packageRules": [
+    {
+      "description": "Keep netlink-bindings and netlink-socket2 updates together",
+      "matchManagers": [
+        "cargo"
+      ],
+      "matchPackageNames": [
+        "netlink-bindings",
+        "netlink-socket2"
+      ],
+      "groupName": "netlink dependencies"
+    }
   ]
 }

src/baseliner.rs

@@ -9,7 +9,6 @@ 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;
@@ -79,7 +78,6 @@ pub struct Baseliner {
     pub selection_tx: Option<Sender<ControlSnapshot>>,
     pub baseline_metrics: MetricsSender,
     pub event_metrics: MetricsSender,
-    pub trace: Recorder,
 }
 
 fn ewma_factor(tick: f64, dur: f64) -> f64 {
@@ -209,23 +207,10 @@ impl Baseliner {
             );
 
             let snapshot = control_snapshot(&reflectors, Instant::now());
-            self.trace.linearize(|trace| {
+            if let Some(selection_tx) = &self.selection_tx {
-                if trace.is_enabled() {
+                let _ = selection_tx.send(snapshot.clone());
-                    trace.record_at(
+            }
-                        reply.last_receive_time_s,
+            self.control_tx.send(snapshot)?;
-                        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)
-            })?;
         }
     }
 }

src/main.rs

@@ -18,7 +18,6 @@ mod ratecontroller;
 mod reflector_selector;
 mod settings;
 mod time;
-mod trace;
 mod util;
 
 use crate::baseliner::Baseliner;
@@ -31,7 +30,7 @@ use crate::ratecontroller::Ratecontroller;
 use crate::reflector_selector::ReflectorSelector;
 use crate::settings::MeasurementType;
 use crate::settings::Settings;
-use ::log::{debug, info, warn};
+use ::log::{info, warn};
 use flume::RecvTimeoutError;
 use std::collections::HashMap;
 use std::net::IpAddr;
@@ -244,7 +243,6 @@ 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 {
@@ -336,7 +334,6 @@ 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();
@@ -420,7 +417,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         reflector_peers,
         reselect_tx,
         rate_metrics,
-        trace.clone(),
     )?;
 
     let err_tx = error_tx.clone();
@@ -464,8 +460,6 @@ fn run(settings: &Settings) -> anyhow::Result<()> {
         &up_shaper,
     );
 
-    trace.finish();
-
     result
 }
 

src/ratecontroller.rs

@@ -14,7 +14,6 @@ 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};
@@ -146,11 +145,9 @@ pub struct Ratecontroller<S: InterfaceStatsProvider, T: TrafficControlBackend> {
     state_dl: State<T::Handle>,
     state_ul: State<T::Handle>,
     rng: fastrand::Rng,
-    next_evaluation_id: u64,
     stats_provider: S,
     traffic_control: T,
     metrics: MetricsSender,
-    trace: Recorder,
 }
 
 impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
@@ -163,7 +160,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         mut stats_provider: S,
         mut traffic_control: T,
         mut rng: fastrand::Rng,
-        trace: Recorder,
     ) -> anyhow::Result<Self> {
         let dl_shaper =
             traffic_control.find_shaper(settings.network.download_interface.as_str())?;
@@ -182,7 +178,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         let (cur_rx, cur_tx) =
             get_interface_stats(&mut stats_provider, &settings.network.upload_interface)?;
 
-        let controller = Self {
+        Ok(Self {
             settings,
             control_snapshot: None,
             control_snapshot_rx,
@@ -191,32 +187,10 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
             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<()> {
@@ -224,20 +198,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         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,
@@ -252,8 +212,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         Ok(())
     }
 
-    fn calculate_rate(&mut self, direction: Direction, evaluation_id: u64) -> anyhow::Result<()> {
+    fn calculate_rate(&mut self, direction: Direction) -> 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,
@@ -272,20 +231,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
 
         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;
@@ -324,20 +269,11 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                 }
 
                 if state.delta_stat > delay_ms {
-                    let selected_index = self.rng.usize(..state.safe_rates.len());
+                    match state
-                    match state.safe_rates.get(selected_index) {
+                        .safe_rates
+                        .get(self.rng.usize(..state.safe_rates.len()))
+                    {
                         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 => {
@@ -413,9 +349,7 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                     &mut self.stats_provider,
                     &self.settings.network.upload_interface,
                 )?;
-                let evaluation_id = self.next_evaluation_id;
+                self.update_deltas()?;
-                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");
@@ -428,20 +362,6 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                     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,
@@ -458,24 +378,8 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
                     continue;
                 }
 
-                self.calculate_rate(Direction::Down, evaluation_id)?;
+                self.calculate_rate(Direction::Down)?;
-                self.calculate_rate(Direction::Up, evaluation_id)?;
+                self.calculate_rate(Direction::Up)?;
-
-                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
@@ -575,36 +479,18 @@ impl<S: InterfaceStatsProvider, T: TrafficControlBackend> Ratecontroller<S, T> {
         }
     }
 
-    fn update_deltas(&mut self, evaluation_id: u64, loop_time: Instant) -> anyhow::Result<()> {
+    fn update_deltas(&mut self) -> anyhow::Result<()> {
-        let recorder = self.trace.clone();
+        drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
-        recorder.linearize(|trace| -> anyhow::Result<()> {
+        let now_t = Instant::now();
-            drain_latest_snapshot(&self.control_snapshot_rx, &mut self.control_snapshot);
+        let reflectors = self.reflectors_lock.read_anyhow()?;
-            let now_t = Instant::now();
+        let (down_deltas, up_deltas) = deltas_from_snapshot(
-            let reflectors = self.reflectors_lock.read_anyhow()?;
+            self.control_snapshot.as_ref(),
-            let (down_deltas, up_deltas) = deltas_from_snapshot(
+            &reflectors,
-                self.control_snapshot.as_ref(),
+            now_t,
-                &reflectors,
+            self.settings.advanced_settings.tick_interval,
-                now_t,
+        );
-                self.settings.advanced_settings.tick_interval,
+        self.state_dl.deltas = down_deltas;
-            );
+        self.state_ul.deltas = up_deltas;
-            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
@@ -627,7 +513,6 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
         reflectors_lock: ArcRwLock<Vec<IpAddr>>,
         reselect_trigger: Sender<bool>,
         metrics: MetricsSender,
-        trace: Recorder,
     ) -> anyhow::Result<Self> {
         Self::new_with_backends(
             settings,
@@ -638,7 +523,6 @@ impl Ratecontroller<PlatformInterfaceStats, PlatformTrafficControl> {
             interface_stats_provider(),
             traffic_control_backend(),
             fastrand::Rng::new(),
-            trace,
         )
     }
 }

src/trace/characterization.rs

@@ -1,497 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use super::reader::{TraceCaptureV1, TraceHeaderV1};
-use super::v1::{Direction, Event, RateRequestReason, Record};
-use std::mem;
-use thiserror::Error;
-
-#[derive(Clone, Debug)]
-pub(super) struct CharacterizationV1 {
-    pub(super) header: TraceHeaderV1,
-    pub(super) initialization: Record,
-    pub(super) startup_rates: Record,
-    pub(super) evaluations: Vec<EvaluationCaseV1>,
-    pub(super) trailing_ping_replies: Vec<Record>,
-}
-
-#[derive(Clone, Debug)]
-pub(super) struct EvaluationCaseV1 {
-    pub(super) ping_replies_before: Vec<Record>,
-    pub(super) boundary: Record,
-    pub(super) download_calculation: Option<Record>,
-    pub(super) upload_calculation: Option<Record>,
-    pub(super) random_choices: Vec<Record>,
-    pub(super) expected_rates: Record,
-    pub(super) deferred_ping_replies: Vec<Record>,
-}
-
-#[derive(Debug, Error)]
-pub(super) enum CharacterizationErrorV1 {
-    #[error("trace has no controller initialization")]
-    MissingInitialization,
-    #[error("trace has no startup requested-rates record")]
-    MissingStartupRates,
-    #[error("record {seq} is structurally unexpected: {detail}")]
-    UnexpectedRecord { seq: u64, detail: &'static str },
-    #[error("evaluation {evaluation_id} is incomplete at EOF")]
-    IncompleteEvaluation { evaluation_id: u64 },
-    #[error("evaluation {evaluation_id} has invalid structure: {detail}")]
-    InvalidEvaluation {
-        evaluation_id: u64,
-        detail: &'static str,
-    },
-}
-
-struct OpenEvaluation {
-    evaluation_id: u64,
-    ping_replies_before: Vec<Record>,
-    boundary: Record,
-    download_calculation: Option<Record>,
-    upload_calculation: Option<Record>,
-    random_choices: Vec<Record>,
-    deferred_ping_replies: Vec<Record>,
-}
-
-pub(super) fn build_characterization_v1(
-    capture: &TraceCaptureV1,
-) -> Result<CharacterizationV1, CharacterizationErrorV1> {
-    let mut initialization = None;
-    let mut startup_rates = None;
-    let mut evaluations = Vec::new();
-    let mut pending_ping_replies = Vec::new();
-    let mut open = None::<OpenEvaluation>;
-
-    for record in &capture.records {
-        match &record.event {
-            Event::PingReply { .. } => {
-                if let Some(evaluation) = &mut open {
-                    evaluation.deferred_ping_replies.push(record.clone());
-                } else {
-                    pending_ping_replies.push(record.clone());
-                }
-            }
-            Event::ControllerInitialized { .. } => {
-                if initialization.is_some() || open.is_some() || !evaluations.is_empty() {
-                    return unexpected(record, "duplicate or late controller initialization");
-                }
-                initialization = Some(record.clone());
-            }
-            Event::ControlEvaluation { evaluation_id, .. } => {
-                if initialization.is_none() || startup_rates.is_none() {
-                    return unexpected(record, "evaluation precedes controller startup");
-                }
-                if open.is_some() {
-                    return unexpected(record, "evaluation begins before the previous one ends");
-                }
-                if evaluations
-                    .last()
-                    .is_none_or(|previous: &EvaluationCaseV1| {
-                        evaluation_id_of(&previous.boundary).wrapping_add(1) != *evaluation_id
-                    })
-                    && (!evaluations.is_empty() || *evaluation_id != 0)
-                {
-                    return invalid(*evaluation_id, "evaluation ids are not contiguous");
-                }
-                open = Some(OpenEvaluation {
-                    evaluation_id: *evaluation_id,
-                    ping_replies_before: mem::take(&mut pending_ping_replies),
-                    boundary: record.clone(),
-                    download_calculation: None,
-                    upload_calculation: None,
-                    random_choices: Vec::new(),
-                    deferred_ping_replies: Vec::new(),
-                });
-            }
-            Event::RateCalculation {
-                evaluation_id,
-                direction,
-            } => {
-                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
-                let calculation = match direction {
-                    Direction::Download => &mut evaluation.download_calculation,
-                    Direction::Upload => &mut evaluation.upload_calculation,
-                };
-                if calculation.replace(record.clone()).is_some() {
-                    return invalid(*evaluation_id, "duplicate directional calculation");
-                }
-            }
-            Event::RandomSafeRateChoice {
-                evaluation_id,
-                direction,
-                ..
-            } => {
-                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
-                let has_calculation = match direction {
-                    Direction::Download => evaluation.download_calculation.is_some(),
-                    Direction::Upload => evaluation.upload_calculation.is_some(),
-                };
-                if !has_calculation {
-                    return invalid(*evaluation_id, "random choice precedes its calculation");
-                }
-                evaluation.random_choices.push(record.clone());
-            }
-            Event::RequestedRates {
-                evaluation_id: None,
-                reason: RateRequestReason::Startup,
-                ..
-            } => {
-                if startup_rates.is_some() || open.is_some() || !evaluations.is_empty() {
-                    return unexpected(record, "duplicate or late startup rates");
-                }
-                startup_rates = Some(record.clone());
-            }
-            Event::RequestedRates {
-                evaluation_id: Some(evaluation_id),
-                reason,
-                ..
-            } => {
-                let evaluation = matching_open(&mut open, record, *evaluation_id)?;
-                match reason {
-                    RateRequestReason::Control
-                        if evaluation.download_calculation.is_none()
-                            || evaluation.upload_calculation.is_none() =>
-                    {
-                        return invalid(*evaluation_id, "control evaluation lacks a calculation");
-                    }
-                    RateRequestReason::NoReflectorData
-                        if evaluation.download_calculation.is_some()
-                            || evaluation.upload_calculation.is_some()
-                            || !evaluation.random_choices.is_empty() =>
-                    {
-                        return invalid(
-                            *evaluation_id,
-                            "no-reflector evaluation contains calculation metadata",
-                        );
-                    }
-                    RateRequestReason::Startup => {
-                        return invalid(*evaluation_id, "startup rates have an evaluation id");
-                    }
-                    _ => {}
-                }
-                let evaluation = open.take().expect("matching evaluation checked above");
-                evaluations.push(EvaluationCaseV1 {
-                    ping_replies_before: evaluation.ping_replies_before,
-                    boundary: evaluation.boundary,
-                    download_calculation: evaluation.download_calculation,
-                    upload_calculation: evaluation.upload_calculation,
-                    random_choices: evaluation.random_choices,
-                    expected_rates: record.clone(),
-                    deferred_ping_replies: evaluation.deferred_ping_replies,
-                });
-            }
-            Event::RequestedRates { .. } => {
-                return unexpected(record, "requested-rates reason and evaluation id disagree");
-            }
-            Event::Header { .. } | Event::End { .. } => {
-                return unexpected(record, "header or end leaked from the trace reader");
-            }
-        }
-    }
-
-    if let Some(evaluation) = open {
-        return Err(CharacterizationErrorV1::IncompleteEvaluation {
-            evaluation_id: evaluation.evaluation_id,
-        });
-    }
-
-    Ok(CharacterizationV1 {
-        header: capture.header.clone(),
-        initialization: initialization.ok_or(CharacterizationErrorV1::MissingInitialization)?,
-        startup_rates: startup_rates.ok_or(CharacterizationErrorV1::MissingStartupRates)?,
-        evaluations,
-        trailing_ping_replies: pending_ping_replies,
-    })
-}
-
-fn matching_open<'a>(
-    open: &'a mut Option<OpenEvaluation>,
-    record: &Record,
-    evaluation_id: u64,
-) -> Result<&'a mut OpenEvaluation, CharacterizationErrorV1> {
-    let Some(evaluation) = open else {
-        return unexpected(record, "evaluation-scoped record has no open evaluation");
-    };
-    if evaluation.evaluation_id != evaluation_id {
-        return invalid(
-            evaluation_id,
-            "record refers to a different open evaluation",
-        );
-    }
-    Ok(evaluation)
-}
-
-fn evaluation_id_of(record: &Record) -> u64 {
-    let Event::ControlEvaluation { evaluation_id, .. } = record.event else {
-        unreachable!("builder only stores control-evaluation boundaries")
-    };
-    evaluation_id
-}
-
-fn unexpected<T>(record: &Record, detail: &'static str) -> Result<T, CharacterizationErrorV1> {
-    Err(CharacterizationErrorV1::UnexpectedRecord {
-        seq: record.seq,
-        detail,
-    })
-}
-
-fn invalid<T>(evaluation_id: u64, detail: &'static str) -> Result<T, CharacterizationErrorV1> {
-    Err(CharacterizationErrorV1::InvalidEvaluation {
-        evaluation_id,
-        detail,
-    })
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use crate::trace::reader::{CaptureCompletionV1, read_trace_v1};
-    use crate::trace::v1::{RateRequestReason, Timestamp};
-    use std::collections::HashSet;
-    use std::io::Cursor;
-
-    const LIVE_FIXTURE: &str = include_str!("../../tests/fixtures/trace-v1/speedtests.jsonl");
-
-    fn record(seq: u64, event: Event) -> Record {
-        Record {
-            seq,
-            timestamp: Timestamp {
-                mono_ns: seq,
-                unix_us: seq as i64,
-            },
-            event,
-        }
-    }
-
-    fn jsonl(records: &[Record]) -> String {
-        records
-            .iter()
-            .map(|record| format!("{}\n", serde_json::to_string(record).unwrap()))
-            .collect()
-    }
-
-    fn initialization(seq: u64) -> Record {
-        record(
-            seq,
-            Event::ControllerInitialized {
-                previous_rx_bytes: 10,
-                previous_tx_bytes: 20,
-                download_prev_mono_ns: 1,
-                upload_prev_mono_ns: 2,
-                download_history_index: 0,
-                upload_history_index: 0,
-                download_safe_rates_kbit: vec![100.0],
-                upload_safe_rates_kbit: vec![200.0],
-            },
-        )
-    }
-
-    fn requested(seq: u64, evaluation_id: Option<u64>, reason: RateRequestReason) -> Record {
-        record(
-            seq,
-            Event::RequestedRates {
-                evaluation_id,
-                reason,
-                download_kbit: 100,
-                upload_kbit: 200,
-                download_requested: true,
-                upload_requested: true,
-            },
-        )
-    }
-
-    fn ping(seq: u64, reflector: &str) -> Record {
-        record(
-            seq,
-            Event::PingReply {
-                reflector: reflector.parse().unwrap(),
-                down_time_ms: 1.0,
-                up_time_ms: 2.0,
-            },
-        )
-    }
-
-    fn boundary(seq: u64, evaluation_id: u64) -> Record {
-        record(
-            seq,
-            Event::ControlEvaluation {
-                evaluation_id,
-                loop_mono_ns: seq,
-                rx_bytes: 30,
-                tx_bytes: 40,
-                active_reflectors: vec!["192.0.2.1".parse().unwrap()],
-            },
-        )
-    }
-
-    fn calculation(seq: u64, evaluation_id: u64, direction: Direction) -> Record {
-        record(
-            seq,
-            Event::RateCalculation {
-                evaluation_id,
-                direction,
-            },
-        )
-    }
-
-    fn synthetic_capture(records: &[Record]) -> TraceCaptureV1 {
-        let mut all = vec![record(0, crate::trace::v1::tests::header_event())];
-        all.extend_from_slice(records);
-        let input = jsonl(&all);
-        read_trace_v1(Cursor::new(input)).unwrap()
-    }
-
-    #[test]
-    fn freezes_evaluation_before_interleaved_ping_replies() {
-        let capture = synthetic_capture(&[
-            initialization(1),
-            requested(2, None, RateRequestReason::Startup),
-            ping(3, "192.0.2.1"),
-            boundary(4, 0),
-            calculation(5, 0, Direction::Download),
-            ping(6, "192.0.2.2"),
-            calculation(7, 0, Direction::Upload),
-            requested(8, Some(0), RateRequestReason::Control),
-            ping(9, "192.0.2.3"),
-            boundary(10, 1),
-            calculation(11, 1, Direction::Download),
-            calculation(12, 1, Direction::Upload),
-            requested(13, Some(1), RateRequestReason::Control),
-        ]);
-
-        let characterization = build_characterization_v1(&capture).unwrap();
-
-        assert_eq!(characterization.header.seq, 0);
-        assert!(characterization.trailing_ping_replies.is_empty());
-        assert_eq!(
-            characterization.evaluations[0].ping_replies_before[0].seq,
-            3
-        );
-        assert_eq!(
-            characterization.evaluations[0].deferred_ping_replies[0].seq,
-            6
-        );
-        assert_eq!(
-            characterization.evaluations[1].ping_replies_before[0].seq,
-            9
-        );
-        assert!(
-            characterization.evaluations[1]
-                .deferred_ping_replies
-                .is_empty()
-        );
-    }
-
-    #[test]
-    fn live_fixture_parses_as_an_incomplete_observed_prefix() {
-        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
-
-        assert_eq!(capture.completion, CaptureCompletionV1::IncompleteEof);
-        assert_eq!(capture.header.program_version, "0.4.1");
-        assert_eq!(capture.records.last().unwrap().seq, 1515);
-    }
-
-    #[test]
-    fn live_fixture_covers_useful_legacy_decisions() {
-        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
-        let characterization = build_characterization_v1(&capture).unwrap();
-
-        assert!(matches!(
-            characterization.initialization.event,
-            Event::ControllerInitialized { .. }
-        ));
-        assert!(matches!(
-            characterization.startup_rates.event,
-            Event::RequestedRates {
-                reason: RateRequestReason::Startup,
-                ..
-            }
-        ));
-        assert!(characterization.evaluations.iter().any(|evaluation| {
-            let Event::ControlEvaluation {
-                active_reflectors, ..
-            } = &evaluation.boundary.event
-            else {
-                unreachable!();
-            };
-            active_reflectors.len() > 5
-        }));
-
-        let stochastic_index = characterization
-            .evaluations
-            .iter()
-            .position(|evaluation| !evaluation.random_choices.is_empty())
-            .expect("fixture must exercise the legacy stochastic branch");
-        let stochastic = &characterization.evaluations[stochastic_index];
-        assert_eq!(evaluation_id_of(&stochastic.boundary), 150);
-        assert!(matches!(
-            stochastic.random_choices.as_slice(),
-            [Record {
-                event: Event::RandomSafeRateChoice {
-                    direction: Direction::Download,
-                    index: 92,
-                    ..
-                },
-                ..
-            }]
-        ));
-        let Event::RequestedRates {
-            download_kbit,
-            download_requested,
-            ..
-        } = stochastic.expected_rates.event
-        else {
-            unreachable!();
-        };
-        assert_eq!(download_kbit, 256_374);
-        assert!(download_requested);
-        let Event::RequestedRates {
-            download_kbit: previous_download_kbit,
-            ..
-        } = characterization.evaluations[stochastic_index - 1]
-            .expected_rates
-            .event
-        else {
-            unreachable!();
-        };
-        assert!(download_kbit < previous_download_kbit);
-
-        let distinct_rates = characterization
-            .evaluations
-            .iter()
-            .map(|evaluation| match evaluation.expected_rates.event {
-                Event::RequestedRates {
-                    download_kbit,
-                    upload_kbit,
-                    ..
-                } => (download_kbit, upload_kbit),
-                _ => unreachable!(),
-            })
-            .collect::<HashSet<_>>();
-        assert!(distinct_rates.len() > 3);
-    }
-
-    #[test]
-    fn live_expected_rates_have_one_unambiguous_evaluation() {
-        let capture = read_trace_v1(Cursor::new(LIVE_FIXTURE)).unwrap();
-        let characterization = build_characterization_v1(&capture).unwrap();
-        let mut ids = HashSet::new();
-
-        for evaluation in &characterization.evaluations {
-            let evaluation_id = evaluation_id_of(&evaluation.boundary);
-            let Event::RequestedRates {
-                evaluation_id: Some(expected_id),
-                reason: RateRequestReason::Control,
-                ..
-            } = evaluation.expected_rates.event
-            else {
-                panic!("fixture evaluation must have normal control output");
-            };
-            assert_eq!(expected_id, evaluation_id);
-            assert!(ids.insert(evaluation_id));
-            assert!(evaluation.download_calculation.is_some());
-            assert!(evaluation.upload_calculation.is_some());
-        }
-    }
-}

src/trace/mod.rs

@@ -1,422 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-pub(crate) mod v1;
-
-#[cfg(all(feature = "trace", test))]
-mod characterization;
-#[cfg(all(feature = "trace", test))]
-mod reader;
-
-#[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<Arc<Inner>>,
-}
-
-#[cfg(feature = "trace")]
-struct Inner {
-    origin: Instant,
-    records: Sender<v1::Record>,
-    next_seq: Mutex<u64>,
-    healthy: Arc<AtomicBool>,
-    writer: Mutex<Option<JoinHandle<()>>>,
-}
-
-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<u64> {
-        self.linearize(|recorder| recorder.record_at(instant, event))
-    }
-
-    pub(crate) fn linearize<R>(&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<W>(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<u64> {
-        #[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<usize> {
-                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| super::reader::parse_record_v1(line).unwrap())
-                .collect::<Vec<_>>();
-
-            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();
-        }
-    }
-}

src/trace/reader.rs

@@ -1,343 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use super::v1::{self, ControlConfig, Event, Record, Timestamp};
-use std::io::{self, BufRead};
-use thiserror::Error;
-
-#[derive(Clone, Debug, PartialEq)]
-pub(super) struct TraceHeaderV1 {
-    pub(super) seq: u64,
-    pub(super) timestamp: Timestamp,
-    pub(super) program_version: String,
-    pub(super) control: ControlConfig,
-}
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub(super) enum CaptureCompletionV1 {
-    Complete,
-    IncompleteEof,
-    IncompleteEnd,
-    TruncatedFinalLine { line: usize },
-}
-
-#[derive(Clone, Debug, PartialEq)]
-pub(super) struct TraceCaptureV1 {
-    pub(super) header: TraceHeaderV1,
-    pub(super) records: Vec<Record>,
-    pub(super) completion: CaptureCompletionV1,
-}
-
-#[derive(Debug, Error)]
-pub(super) enum TraceReadErrorV1 {
-    #[error("failed to read trace line {line}: {source}")]
-    Io {
-        line: usize,
-        #[source]
-        source: io::Error,
-    },
-    #[error("trace line {line} is not a valid v1 record: {source}")]
-    MalformedRecord {
-        line: usize,
-        #[source]
-        source: serde_json::Error,
-    },
-    #[error("trace is missing its v1 header")]
-    MissingHeader,
-    #[error("first trace record must be the v1 header")]
-    FirstRecordNotHeader,
-    #[error("trace header sequence must be 0, got {actual}")]
-    InvalidHeaderSequence { actual: u64 },
-    #[error("unsupported trace format {actual:?}")]
-    InvalidFormat { actual: String },
-    #[error("unsupported trace version {actual}")]
-    InvalidVersion { actual: u16 },
-    #[error("trace line {line} has sequence {actual}, expected {expected}")]
-    InvalidSequence {
-        line: usize,
-        expected: u64,
-        actual: u64,
-    },
-    #[error("trace line {line} contains another header")]
-    UnexpectedHeader { line: usize },
-    #[error("trace line {line} follows an end record")]
-    RecordAfterEnd { line: usize },
-}
-
-pub(super) fn read_trace_v1(mut reader: impl BufRead) -> Result<TraceCaptureV1, TraceReadErrorV1> {
-    let mut header = None;
-    let mut records = Vec::new();
-    let mut next_seq = 0;
-    let mut completion = None;
-    let mut line_number = 0;
-
-    loop {
-        let mut line = String::new();
-        line_number += 1;
-        let bytes_read = reader
-            .read_line(&mut line)
-            .map_err(|source| TraceReadErrorV1::Io {
-                line: line_number,
-                source,
-            })?;
-        if bytes_read == 0 {
-            break;
-        }
-
-        if completion.is_some() {
-            return Err(TraceReadErrorV1::RecordAfterEnd { line: line_number });
-        }
-
-        let terminated = line.ends_with('\n');
-        let json = line.strip_suffix('\n').unwrap_or(&line);
-        let json = json.strip_suffix('\r').unwrap_or(json);
-        let record = match parse_record_v1(json) {
-            Ok(record) => record,
-            Err(source)
-                if !terminated
-                    && !json.trim().is_empty()
-                    && source.is_eof()
-                    && header.is_some() =>
-            {
-                return Ok(TraceCaptureV1 {
-                    header: header.expect("header checked above"),
-                    records,
-                    completion: CaptureCompletionV1::TruncatedFinalLine { line: line_number },
-                });
-            }
-            Err(source) => {
-                return Err(TraceReadErrorV1::MalformedRecord {
-                    line: line_number,
-                    source,
-                });
-            }
-        };
-
-        if record.seq != next_seq {
-            if header.is_none() && next_seq == 0 {
-                return Err(TraceReadErrorV1::InvalidHeaderSequence { actual: record.seq });
-            }
-            return Err(TraceReadErrorV1::InvalidSequence {
-                line: line_number,
-                expected: next_seq,
-                actual: record.seq,
-            });
-        }
-        next_seq = next_seq.saturating_add(1);
-
-        if header.is_none() {
-            let Event::Header {
-                format,
-                version,
-                program_version,
-                control,
-            } = record.event
-            else {
-                return Err(TraceReadErrorV1::FirstRecordNotHeader);
-            };
-            if format != v1::FORMAT_NAME {
-                return Err(TraceReadErrorV1::InvalidFormat { actual: format });
-            }
-            if version != v1::FORMAT_VERSION {
-                return Err(TraceReadErrorV1::InvalidVersion { actual: version });
-            }
-            header = Some(TraceHeaderV1 {
-                seq: record.seq,
-                timestamp: record.timestamp,
-                program_version,
-                control,
-            });
-            continue;
-        }
-
-        match record.event {
-            Event::Header { .. } => {
-                return Err(TraceReadErrorV1::UnexpectedHeader { line: line_number });
-            }
-            Event::End { clean } => {
-                completion = Some(if clean {
-                    CaptureCompletionV1::Complete
-                } else {
-                    CaptureCompletionV1::IncompleteEnd
-                });
-            }
-            _ => records.push(record),
-        }
-    }
-
-    let header = header.ok_or(TraceReadErrorV1::MissingHeader)?;
-    Ok(TraceCaptureV1 {
-        header,
-        records,
-        completion: completion.unwrap_or(CaptureCompletionV1::IncompleteEof),
-    })
-}
-
-pub(super) fn parse_record_v1(json: &str) -> Result<Record, serde_json::Error> {
-    serde_json::from_str(json)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use std::io::Cursor;
-
-    fn record(seq: u64, event: Event) -> Record {
-        Record {
-            seq,
-            timestamp: Timestamp {
-                mono_ns: seq,
-                unix_us: seq as i64,
-            },
-            event,
-        }
-    }
-
-    fn jsonl(records: &[Record]) -> String {
-        records
-            .iter()
-            .map(|record| format!("{}\n", serde_json::to_string(record).unwrap()))
-            .collect()
-    }
-
-    fn header(seq: u64) -> Record {
-        record(seq, v1::tests::header_event())
-    }
-
-    fn ping(seq: u64) -> Record {
-        record(
-            seq,
-            Event::PingReply {
-                reflector: "192.0.2.1".parse().unwrap(),
-                down_time_ms: 1.0,
-                up_time_ms: 2.0,
-            },
-        )
-    }
-
-    #[test]
-    fn parses_clean_v1_trace() {
-        let input = jsonl(&[header(0), ping(1), record(2, Event::End { clean: true })]);
-
-        let capture = read_trace_v1(Cursor::new(input)).unwrap();
-
-        assert_eq!(capture.header.seq, 0);
-        assert_eq!(capture.records, vec![ping(1)]);
-        assert_eq!(capture.completion, CaptureCompletionV1::Complete);
-    }
-
-    #[test]
-    fn retains_records_before_truncated_final_line() {
-        let input = format!(
-            "{}{{\"seq\":2,\"timestamp\":{{",
-            jsonl(&[header(0), ping(1)])
-        );
-
-        let capture = read_trace_v1(Cursor::new(input)).unwrap();
-
-        assert_eq!(capture.records, vec![ping(1)]);
-        assert_eq!(
-            capture.completion,
-            CaptureCompletionV1::TruncatedFinalLine { line: 3 }
-        );
-    }
-
-    #[test]
-    fn marks_valid_eof_without_end_incomplete() {
-        let capture = read_trace_v1(Cursor::new(jsonl(&[header(0), ping(1)]))).unwrap();
-
-        assert_eq!(capture.records, vec![ping(1)]);
-        assert_eq!(capture.completion, CaptureCompletionV1::IncompleteEof);
-    }
-
-    #[test]
-    fn rejects_malformed_middle_line() {
-        let input = format!("{}not-json\n{}", jsonl(&[header(0)]), jsonl(&[ping(1)]));
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(input)),
-            Err(TraceReadErrorV1::MalformedRecord { line: 2, .. })
-        ));
-    }
-
-    #[test]
-    fn rejects_sequence_gap_duplicate_and_out_of_order() {
-        let cases = [
-            (vec![header(0), ping(2)], 2, 1, 2),
-            (vec![header(0), ping(1), ping(1)], 3, 2, 1),
-            (vec![header(0), ping(1), ping(2), ping(1)], 4, 3, 1),
-        ];
-
-        for (records, line, expected, actual) in cases {
-            let input = jsonl(&records);
-            assert!(matches!(
-                read_trace_v1(Cursor::new(input)),
-                Err(TraceReadErrorV1::InvalidSequence {
-                    line: seen_line,
-                    expected: seen_expected,
-                    actual: seen,
-                }) if seen_line == line && seen_expected == expected && seen == actual
-            ));
-        }
-    }
-
-    #[test]
-    fn rejects_non_truncation_error_on_final_line() {
-        let input = format!("{}{{\"seq\":1,]", jsonl(&[header(0)]));
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(input)),
-            Err(TraceReadErrorV1::MalformedRecord { line: 2, .. })
-        ));
-    }
-
-    #[test]
-    fn rejects_records_after_clean_end() {
-        let input = jsonl(&[header(0), record(1, Event::End { clean: true }), ping(2)]);
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(input)),
-            Err(TraceReadErrorV1::RecordAfterEnd { line: 3 })
-        ));
-    }
-
-    #[test]
-    fn validates_format_name_and_version() {
-        let Event::Header {
-            program_version,
-            control,
-            ..
-        } = v1::tests::header_event()
-        else {
-            unreachable!();
-        };
-        let invalid_format = record(
-            0,
-            Event::Header {
-                format: "other".to_owned(),
-                version: v1::FORMAT_VERSION,
-                program_version: program_version.clone(),
-                control: control.clone(),
-            },
-        );
-        let invalid_version = record(
-            0,
-            Event::Header {
-                format: v1::FORMAT_NAME.to_owned(),
-                version: v1::FORMAT_VERSION + 1,
-                program_version,
-                control,
-            },
-        );
-
-        assert!(matches!(
-            read_trace_v1(Cursor::new(jsonl(&[invalid_format]))),
-            Err(TraceReadErrorV1::InvalidFormat { .. })
-        ));
-        assert!(matches!(
-            read_trace_v1(Cursor::new(jsonl(&[invalid_version]))),
-            Err(TraceReadErrorV1::InvalidVersion { .. })
-        ));
-    }
-}

src/trace/v1.rs

@@ -1,269 +0,0 @@
-// SPDX-FileCopyrightText: 2026-Present Nils Andreas Svee mailto:contact@lochnair.net (github @Lochnair)
-//
-// SPDX-License-Identifier: MPL-2.0
-
-use serde::de::{Deserializer, Visitor};
-use serde::{Deserialize, Serialize};
-use std::fmt;
-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 {
-        #[serde(deserialize_with = "deserialize_i128")]
-        previous_rx_bytes: i128,
-        #[serde(deserialize_with = "deserialize_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<f64>,
-        upload_safe_rates_kbit: Vec<f64>,
-    },
-    ControlEvaluation {
-        evaluation_id: u64,
-        loop_mono_ns: u64,
-        #[serde(deserialize_with = "deserialize_i128")]
-        rx_bytes: i128,
-        #[serde(deserialize_with = "deserialize_i128")]
-        tx_bytes: i128,
-        active_reflectors: Vec<IpAddr>,
-    },
-    RateCalculation {
-        evaluation_id: u64,
-        direction: Direction,
-    },
-    RandomSafeRateChoice {
-        evaluation_id: u64,
-        direction: Direction,
-        index: u64,
-        rate_kbit: f64,
-    },
-    RequestedRates {
-        evaluation_id: Option<u64>,
-        reason: RateRequestReason,
-        download_kbit: u64,
-        upload_kbit: u64,
-        download_requested: bool,
-        upload_requested: bool,
-    },
-    End {
-        clean: bool,
-    },
-}
-
-fn deserialize_i128<'de, D>(deserializer: D) -> Result<i128, D::Error>
-where
-    D: Deserializer<'de>,
-{
-    struct I128Visitor;
-
-    impl Visitor<'_> for I128Visitor {
-        type Value = i128;
-
-        fn expecting(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
-            formatter.write_str("a signed or unsigned integer")
-        }
-
-        fn visit_i64<E>(self, value: i64) -> Result<Self::Value, E> {
-            Ok(i128::from(value))
-        }
-
-        fn visit_u64<E>(self, value: u64) -> Result<Self::Value, E> {
-            Ok(i128::from(value))
-        }
-    }
-
-    deserializer.deserialize_any(I128Visitor)
-}
-
-#[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!(
-            crate::trace::reader::parse_record_v1(&serde_json::to_string(&record).unwrap())
-                .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);
-    }
-}

src/trace/writer.rs

@@ -1,54 +0,0 @@
-// 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<W>(
-    writer: W,
-    records: Receiver<Record>,
-    header: Record,
-    healthy: Arc<AtomicBool>,
-) -> io::Result<JoinHandle<()>>
-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<W>(mut writer: W, records: Receiver<Record>, 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")
-}

tests/fixtures/trace-v1/README.md

@@ -1,7 +0,0 @@
-# Trace v1 characterization fixture
-
-`speedtests.jsonl` is an observed prefix of a live legacy-controller capture
-recorded during speed and LibreQoS testing. It preserves the original records,
-sequence values, and order through evaluation 150. The prefix intentionally has
-no synthetic `end` record and may include undesirable legacy behavior; it is an
-extraction-characterization artifact, not a correctness oracle.