Author SHA1 Message Date
renovate Bot 713f0d8d8b chore(deps): update google/osv-scanner-action action to v2.5.1
Build / build (pull_request) Successful in 1m6s
DevSkim / DevSkim (pull_request) Failing after 1m21s
rust-clippy analyze / Run rust-clippy analyzing (pull_request) Failing after 2m1s
2026-08-29 14:27:18 +00:00
Lochnair 083d464067 Merge pull request 'chore(deps): update netlink dependencies to v0.3.4' (#76) from renovate/netlink-dependencies into master
Build / build (push) Successful in 1m5s
DevSkim / DevSkim (push) Failing after 1m21s
rust-clippy analyze / Run rust-clippy analyzing (push) Failing after 2m1s
Reviewed-on: #76
2026-08-29 14:11:36 +00:00
renovate Bot f0eae4b2ce chore(deps): update netlink dependencies to v0.3.4
Build / build (pull_request) Successful in 1m4s
DevSkim / DevSkim (pull_request) Failing after 1m32s
rust-clippy analyze / Run rust-clippy analyzing (pull_request) Failing after 2m5s
2026-08-29 14:10:14 +00:00
Lochnair a6067910ad chore: make renovate combine netlink deps
Build / build (push) Successful in 1m4s
DevSkim / DevSkim (push) Failing after 1m18s
rust-clippy analyze / Run rust-clippy analyzing (push) Failing after 1m57s
2026-08-28 09:15:20 +00:00
Lochnair ffd7b0beb7 fix: remove unused import
DevSkim / DevSkim (push) Failing after 4s
Build / build (push) Successful in 57s
rust-clippy analyze / Run rust-clippy analyzing (push) Failing after 2m10s
2026-08-23 17:23:10 +00:00
Lochnair 377fb3be2d chore(ci): install latest rust
DevSkim / DevSkim (push) Failing after 4s
rust-clippy analyze / Run rust-clippy analyzing (push) Failing after 1m42s
Build / build (push) Failing after 50s
2026-08-23 17:12:17 +00:00
Lochnair d765bef002 chore(ci): add workflow_dispatch to build workflow
Build / build (push) Failing after 2s
DevSkim / DevSkim (push) Failing after 4s
rust-clippy analyze / Run rust-clippy analyzing (push) Failing after 5s
2026-08-23 16:50:26 +00:00
renovate[bot] 86b13f948e chore(deps): update rust crate thiserror to v2.0.20 2026-08-14 16:44:07 +00:00
12 changed files with 55 additions and 891 deletions

No files matched your search

+4
View File
@@ -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
+2 -2
View File
@@ -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: |-
Generated
+8 -9
View File
@@ -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",
+1 -3
View File
@@ -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"]
-8
View File
@@ -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:
+13
View File
@@ -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"
}
]
}
+4 -19
View File
@@ -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 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)
})?;
if let Some(selection_tx) = &self.selection_tx {
let _ = selection_tx.send(snapshot.clone());
}
self.control_tx.send(snapshot)?;
}
}
}
+1 -7
View File
@@ -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
}
+22 -138
View File
@@ -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<()> {
let trace = self.trace.clone();
fn calculate_rate(&mut self, direction: Direction) -> anyhow::Result<()> {
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.safe_rates.get(selected_index) {
match state
.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.next_evaluation_id = self.next_evaluation_id.wrapping_add(1);
self.update_deltas(evaluation_id, now_t)?;
self.update_deltas()?;
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::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,
},
);
}
self.calculate_rate(Direction::Down)?;
self.calculate_rate(Direction::Up)?;
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<()> {
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(())
})?;
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;
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,
)
}
}
-417
View File
@@ -1,417 +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(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| serde_json::from_str::<v1::Record>(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();
}
}
}
-234
View File
@@ -1,234 +0,0 @@
// 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<f64>,
upload_safe_rates_kbit: Vec<f64>,
},
ControlEvaluation {
evaluation_id: u64,
loop_mono_ns: u64,
rx_bytes: 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,
},
}
#[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::<Record>(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);
}
}
-54
View File
@@ -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")
}