Compare commits
2
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8af8f90200 | ||
|
|
4973cff43e |
No files matched your search
+7634
File diff suppressed because it is too large.
Load diff
@@ -0,0 +1,497 @@
|
||||
// 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());
|
||||
}
|
||||
}
|
||||
}
|
||||
+6
-1
@@ -4,6 +4,11 @@
|
||||
|
||||
pub(crate) mod v1;
|
||||
|
||||
#[cfg(all(feature = "trace", test))]
|
||||
mod characterization;
|
||||
#[cfg(all(feature = "trace", test))]
|
||||
mod reader;
|
||||
|
||||
#[cfg(feature = "trace")]
|
||||
mod writer;
|
||||
|
||||
@@ -374,7 +379,7 @@ mod tests {
|
||||
let contents = std::fs::read_to_string(&path).unwrap();
|
||||
let records = contents
|
||||
.lines()
|
||||
.map(|line| serde_json::from_str::<v1::Record>(line).unwrap())
|
||||
.map(|line| super::reader::parse_record_v1(line).unwrap())
|
||||
.collect::<Vec<_>>();
|
||||
|
||||
assert_eq!(records.len(), 102);
|
||||
|
||||
@@ -0,0 +1,343 @@
|
||||
// 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 { .. })
|
||||
));
|
||||
}
|
||||
}
|
||||
+36
-1
@@ -2,7 +2,9 @@
|
||||
//
|
||||
// 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")]
|
||||
@@ -51,7 +53,9 @@ pub(crate) enum Event {
|
||||
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,
|
||||
@@ -63,7 +67,9 @@ pub(crate) enum Event {
|
||||
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>,
|
||||
},
|
||||
@@ -90,6 +96,31 @@ pub(crate) enum Event {
|
||||
},
|
||||
}
|
||||
|
||||
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 {
|
||||
@@ -218,7 +249,11 @@ pub(super) mod tests {
|
||||
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);
|
||||
assert_eq!(
|
||||
crate::trace::reader::parse_record_v1(&serde_json::to_string(&record).unwrap())
|
||||
.unwrap(),
|
||||
record
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
Vendored
+7
@@ -0,0 +1,7 @@
|
||||
# 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.
|
||||
+1516
File diff suppressed because it is too large.
Load diff
Reference in new issue
Block a user