diff --git a/src/lib.rs b/src/lib.rs index 3775c28..c87d273 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -238,51 +238,76 @@ pub trait DuplicateTrace: Send { fn next_duplicate(&mut self) -> Option<(DuplicatePattern, Duration)>; } -/// The action a rwnd trace instructs the receiver to take at a single step. -/// -/// At most one action is present per step; a step that only reconfigures the -/// receive buffer (`set_rcv_buf`) without any read or observed-remaining update -/// leaves [`RwndDecision::action`] as `None`. -/// -/// - `AppRead` drives the receiver model by simulating the application reading -/// `bytes` from the receive buffer; the resulting rwnd is computed from the -/// buffer state. -/// - `Remaining` skips the simulation and directly enforces an observed rwnd -/// of `rwnd` bytes — useful for replaying captured traces where only the -/// advertised window is known. -#[derive(Debug, Clone, PartialEq)] -pub enum RwndAction { - /// The simulated application reads this many bytes from the receive buffer at this step. - AppRead { bytes: u64 }, - /// The remaining rwnd value observed immediately after the app consumes data at this step. - Remaining { rwnd: u64 }, -} - /// A single receive-side decision emitted by a [`RwndTrace`]. /// -/// Each step of a rwnd trace produces one `RwndDecision` paired with a -/// [`Duration`] (see [`RwndTrace`]). Both fields are optional and independent: -/// a step may resize the socket buffer, advance the receive model, both, or -/// neither (though a step that sets neither is effectively a no-op). -#[derive(Debug, Clone, PartialEq)] +/// The two fields are independent, and between them they say what the receiver +/// does for this step's duration. +/// +/// `set_rcv_buf` sizes the receive buffer. On its own it also states that the +/// application is keeping up with it: the buffer holds that many bytes, the +/// advertised window is held at that value, and the application drains +/// continuously so the window's right edge slides forward with the data +/// received. That models a receiver whose buffer stopped growing -- in-flight +/// ends up limited by the window, and the window itself stays put. +/// +/// `app_read_bytes` states the opposite situation: over this step the +/// application reads exactly that many bytes and then stops. The window is +/// whatever is left of the buffer once the unread backlog is subtracted, so it +/// shrinks as data arrives and reaches zero when the buffer fills. That models a +/// receiver whose application is the bottleneck. +/// +/// Because the fields are independent, all four combinations are meaningful and +/// none is a special case: +/// +/// | `set_rcv_buf` | `app_read_bytes` | meaning | +/// |---|---|---| +/// | `Some(n)` | `None` | buffer `n`, window pinned at `n`, application drains continuously | +/// | `None` | `Some(m)` | read `m` bytes against the standing buffer; window is what is left | +/// | `Some(n)` | `Some(m)` | resize the buffer to `n`, then read `m` bytes from it | +/// | `None` | `None` | carry the previous configuration forward for this step | +/// +/// # Both fields on one step +/// +/// The two apply in order: the buffer is resized first, and the read is taken +/// against the new size. Three consequences are worth stating outright, because +/// they are what distinguishes this from a buffer-only step: +/// +/// - **The window is not pinned.** The "application keeps up" half of +/// `set_rcv_buf` belongs to a step that states no read. Once `app_read_bytes` +/// is present it is the application's behaviour, so the window follows from +/// `n - unread` and decays as data arrives, exactly as for a read-only step. +/// A step of `set_rcv_buf: n` and `app_read_bytes: m` is therefore *not* +/// equivalent to a buffer-only step of `n` followed by a read of `m`. +/// - **The backlog survives the resize.** Only the capacity changes; bytes +/// already received and not yet read stay unread. Resizing does not discard, +/// deliver, or otherwise account for them. +/// - **The window saturates at zero.** With `unread` bytes outstanding the +/// window is `n.saturating_sub(unread)`, so resizing to a value at or below +/// the current backlog advertises a zero window until the application reads +/// its way back under the new size. This is the intended way to state a +/// receiver that shrank its buffer while behind. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub struct RwndDecision { - /// If `Some`, reconfigure the socket's receive buffer to this size at this step. + /// If `Some`, size the receive buffer to this many bytes and hold the + /// advertised window there, with the application draining continuously. pub set_rcv_buf: Option, - /// If `Some`, the app-read or observed-remaining action for this step. - pub action: Option, + /// If `Some`, the application reads exactly this many bytes over this step + /// and then stops; the window follows from what is left unread. + pub app_read_bytes: Option, } /// This is a trait that represents a trace of receive-window decisions over time. /// /// The trace is a sequence of `(rwnd_decision, duration)` pairs. The decision -/// describes how the socket's receive buffer, the application's read behavior, -/// and/or the observed remaining window change at this step; the duration is -/// how long this configuration lasts before the next step applies. +/// describes what the receiver does -- how large its buffer is and how its +/// application reads from it -- and the duration is how long that lasts before +/// the next step applies. /// /// For example, if the sequence is -/// `[(set_rcv_buf=64KB, app_read=1KB, 1s), (rwnd_remaining=32KB, 2s)]`, -/// then the receive buffer is resized to 64KB and the app reads 1KB for 1s, -/// then the observed rwnd becomes 32KB for 2s. +/// `[(set_rcv_buf=64KB, 1s), (app_read_bytes=1KB, 2s)]`, then for 1s the +/// receiver holds a 64KB window and drains it as fast as data arrives, and for +/// the next 2s its application reads only 1KB, so the window decays from 64KB +/// as the unread backlog grows. /// /// Each `next_rwnd` call returns **the next decision and its duration** in the /// sequence, or **None** when the trace is exhausted. Mirrors the shape of diff --git a/src/model/rwnd.rs b/src/model/rwnd.rs index 62506db..abf5ec7 100644 --- a/src/model/rwnd.rs +++ b/src/model/rwnd.rs @@ -7,13 +7,30 @@ //! - [`StaticRwnd`]: A trace model with a single rwnd decision. //! - [`RepeatedRwndPattern`]: A trace model with a repeated rwnd pattern. //! +//! ## Step semantics +//! +//! A step carries two independent fields, `set_rcv_buf` and `app_read_bytes`, +//! and between them they say what the receiver does for the step's duration. +//! See [`RwndDecision`] for the full table; in short, `set_rcv_buf` states a +//! buffer the application keeps drained (so the window stays at that value), +//! and `app_read_bytes` states an application that reads only so much (so the +//! window decays as the backlog grows). +//! +//! A step carrying **both** resizes the buffer first and takes the read against +//! the new size. The window is then `set_rcv_buf - unread`, not pinned at +//! `set_rcv_buf`: stating a read means the application is the bottleneck, so the +//! "application keeps up" half of a buffer-only step no longer applies. The +//! unread backlog carries across the resize untouched, so a resize to at or +//! below the current backlog advertises a zero window until the application +//! catches up. +//! //! ## Examples //! //! An example to build model from configuration: //! //! ``` //! # use netem_trace::model::StaticRwndConfig; -//! # use netem_trace::{Duration, RwndTrace, RwndAction}; +//! # use netem_trace::{Duration, RwndTrace}; //! let mut static_rwnd = StaticRwndConfig::new() //! .set_rcv_buf(65536) //! .app_read(1024) @@ -21,39 +38,70 @@ //! .build(); //! let (decision, duration) = static_rwnd.next_rwnd().unwrap(); //! assert_eq!(decision.set_rcv_buf, Some(65536)); -//! assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +//! assert_eq!(decision.app_read_bytes, Some(1024)); //! assert_eq!(duration, Duration::from_secs(1)); //! assert_eq!(static_rwnd.next_rwnd(), None); //! ``` //! +//! The step above carries both fields, so it resizes the buffer to 64 KiB and +//! then has the application read 1 KiB from it. The window that follows is +//! `65536 - unread` rather than a window pinned at 65536 -- compare the +//! buffer-only step below, which does pin it: +//! +//! ``` +//! # use netem_trace::model::StaticRwndConfig; +//! # use netem_trace::{Duration, RwndTrace}; +//! // Buffer only: the window is held at 65536 and the application keeps up. +//! let mut pinned = StaticRwndConfig::new() +//! .set_rcv_buf(65536) +//! .duration(Duration::from_secs(1)) +//! .build(); +//! let (decision, _) = pinned.next_rwnd().unwrap(); +//! assert_eq!(decision.set_rcv_buf, Some(65536)); +//! assert_eq!(decision.app_read_bytes, None); +//! +//! // Both: same buffer, but the application now reads only 1 KiB per step, so +//! // the window follows the backlog instead of staying at 65536. +//! let mut app_limited = StaticRwndConfig::new() +//! .set_rcv_buf(65536) +//! .app_read(1024) +//! .duration(Duration::from_secs(1)) +//! .build(); +//! let (decision, _) = app_limited.next_rwnd().unwrap(); +//! assert_eq!(decision.set_rcv_buf, Some(65536)); +//! assert_eq!(decision.app_read_bytes, Some(1024)); +//! +//! // Shrinking the buffer under a standing backlog is stated the same way; the +//! // window saturates at zero until the application reads its way back under it. +//! let mut shrink = StaticRwndConfig::new() +//! .set_rcv_buf(8192) +//! .app_read(0) +//! .duration(Duration::from_secs(1)) +//! .build(); +//! let (decision, _) = shrink.next_rwnd().unwrap(); +//! assert_eq!(decision.set_rcv_buf, Some(8192)); +//! assert_eq!(decision.app_read_bytes, Some(0)); +//! ``` +//! //! A more common use case is to build model from a configuration file (e.g. json file): //! //! ``` //! # use netem_trace::model::{StaticRwndConfig, RwndTraceConfig}; -//! # use netem_trace::{Duration, RwndTrace, RwndAction}; +//! # use netem_trace::{Duration, RwndTrace}; //! # #[cfg(feature = "human")] -//! # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":\"1s\",\"rwnd_remaining\":32768}}],\"count\":2}}"; -//! // The content would be "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}],\"count\":2}}" +//! # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536}},{\"StaticRwndConfig\":{\"duration\":\"1s\",\"app_read_bytes\":1024}}],\"count\":2}}"; +//! // The content would be "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"app_read_bytes\":1024}}],\"count\":2}}" //! // if the `human` feature is not enabled. //! # #[cfg(not(feature = "human"))] -//! let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}],\"count\":2}}"; +//! let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"app_read_bytes\":1024}}],\"count\":2}}"; //! let des: Box = serde_json::from_str(config_file_content).unwrap(); //! let mut model = des.into_model(); //! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); -//! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); -//! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +//! assert_eq!(decision.set_rcv_buf, Some(65536)); //! let (decision, _) = model.next_rwnd().unwrap(); -//! assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); -//! assert_eq!(model.next_rwnd(), None); +//! assert_eq!(decision.app_read_bytes, Some(1024)); //! ``` -//! -//! At most one of `app_read_bytes` or `rwnd_remaining` may be set per step — -//! never both. A step with neither produces [`RwndDecision::action`] as `None`, -//! which is valid for steps that only reconfigure the receive buffer. -use crate::{Duration, RwndAction, RwndDecision, RwndTrace}; +use crate::{Duration, RwndDecision, RwndTrace}; use dyn_clone::DynClone; /// This trait is used to convert a rwnd trace configuration into a rwnd trace model. @@ -70,7 +118,7 @@ pub trait RwndTraceConfig: DynClone + Send { dyn_clone::clone_trait_object!(RwndTraceConfig); #[cfg(feature = "serde")] -use serde::{Deserialize, Deserializer, Serialize, Serializer}; +use serde::{Deserialize, Serialize}; /// The model of a static rwnd trace: a single decision valid for one duration. /// @@ -78,15 +126,14 @@ use serde::{Deserialize, Deserializer, Serialize, Serializer}; /// /// ``` /// # use netem_trace::model::StaticRwndConfig; -/// # use netem_trace::{Duration, RwndTrace, RwndAction}; +/// # use netem_trace::{Duration, RwndTrace}; /// let mut static_rwnd = StaticRwndConfig::new() /// .set_rcv_buf(65536) -/// .app_read(1024) /// .duration(Duration::from_secs(1)) /// .build(); /// let (decision, duration) = static_rwnd.next_rwnd().unwrap(); /// assert_eq!(decision.set_rcv_buf, Some(65536)); -/// assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +/// assert_eq!(decision.app_read_bytes, None); /// assert_eq!(duration, Duration::from_secs(1)); /// assert_eq!(static_rwnd.next_rwnd(), None); /// ``` @@ -99,85 +146,39 @@ pub struct StaticRwnd { /// The configuration struct for [`StaticRwnd`]. /// /// The serialized JSON form is **flat**: a step looks like -/// `{"duration":"1s","set_rcv_buf":65536,"app_read_bytes":1024}` (or -/// `{"duration":"1s","rwnd_remaining":32768}`), never with an `action` wrapper. +/// `{"duration":"1s","set_rcv_buf":65536}` or +/// `{"duration":"1s","app_read_bytes":1024}`, and may carry both keys. /// -/// At most one of `app_read_bytes` / `rwnd_remaining` may be set; the deserializer -/// rejects inputs where both are present. A step with neither is valid and produces -/// [`RwndDecision::action`] as `None` (useful for steps that only reconfigure the -/// receive buffer). +/// The two fields are independent -- there is no invariant to enforce, so +/// `Serialize`/`Deserialize` are derived. A step with neither field is valid and +/// carries the previous configuration forward. +/// +/// Unknown keys are rejected rather than ignored. The schema dropped +/// `rwnd_remaining`, and serde's default of skipping what it does not recognise +/// would turn a trace written against the old schema into a run of steps that +/// state nothing -- a replay that looks healthy while reproducing no receiver at +/// all. Failing to deserialize names the offending field instead. +#[cfg_attr( + feature = "serde", + derive(Serialize, Deserialize), + serde(default, deny_unknown_fields) +)] #[derive(Debug, Clone, Default)] pub struct StaticRwndConfig { + #[cfg_attr( + feature = "human", + serde(with = "humantime_serde"), + serde(skip_serializing_if = "Option::is_none") + )] + #[cfg_attr( + all(feature = "serde", not(feature = "human")), + serde(skip_serializing_if = "Option::is_none") + )] pub duration: Option, + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] pub set_rcv_buf: Option, - pub action: Option, -} - -#[cfg(feature = "serde")] -impl<'de> Deserialize<'de> for StaticRwndConfig { - fn deserialize>(deserializer: D) -> Result { - #[derive(Deserialize, Default)] - #[serde(default)] - struct Helper { - #[cfg_attr(feature = "human", serde(with = "humantime_serde"))] - #[serde(default)] - duration: Option, - #[serde(default)] - set_rcv_buf: Option, - #[serde(default)] - app_read_bytes: Option, - #[serde(default)] - rwnd_remaining: Option, - } - - let h = Helper::deserialize(deserializer)?; - let action = match (h.app_read_bytes, h.rwnd_remaining) { - (Some(bytes), None) => Some(RwndAction::AppRead { bytes }), - (None, Some(rwnd)) => Some(RwndAction::Remaining { rwnd }), - (Some(_), Some(_)) => { - return Err(serde::de::Error::custom( - "rwnd step cannot set both `app_read_bytes` and `rwnd_remaining`", - )); - } - (None, None) => None, - }; - Ok(Self { - duration: h.duration, - set_rcv_buf: h.set_rcv_buf, - action, - }) - } -} - -#[cfg(feature = "serde")] -impl Serialize for StaticRwndConfig { - fn serialize(&self, serializer: S) -> Result { - #[derive(Serialize)] - struct Out { - #[serde(skip_serializing_if = "Option::is_none")] - #[cfg_attr(feature = "human", serde(with = "humantime_serde"))] - duration: Option, - #[serde(skip_serializing_if = "Option::is_none")] - set_rcv_buf: Option, - #[serde(skip_serializing_if = "Option::is_none")] - app_read_bytes: Option, - #[serde(skip_serializing_if = "Option::is_none")] - rwnd_remaining: Option, - } - - let (app_read_bytes, rwnd_remaining) = match &self.action { - Some(RwndAction::AppRead { bytes }) => (Some(*bytes), None), - Some(RwndAction::Remaining { rwnd }) => (None, Some(*rwnd)), - None => (None, None), - }; - Out { - duration: self.duration, - set_rcv_buf: self.set_rcv_buf, - app_read_bytes, - rwnd_remaining, - } - .serialize(serializer) - } + #[cfg_attr(feature = "serde", serde(skip_serializing_if = "Option::is_none"))] + pub app_read_bytes: Option, } /// The model contains an array of rwnd trace models. @@ -194,15 +195,16 @@ impl Serialize for StaticRwndConfig { /// /// ``` /// # use netem_trace::model::{StaticRwndConfig, RwndTraceConfig}; -/// # use netem_trace::{Duration, RwndTrace, RwndAction}; +/// # use netem_trace::{Duration, RwndTrace}; /// # #[cfg(feature = "human")] -/// # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":\"1s\",\"rwnd_remaining\":32768}}],\"count\":2}}"; +/// # let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}],\"count\":2}}"; /// # #[cfg(not(feature = "human"))] -/// let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}},{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}],\"count\":2}}"; +/// let config_file_content = "{\"RepeatedRwndPatternConfig\":{\"pattern\":[{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}],\"count\":2}}"; /// let des: Box = serde_json::from_str(config_file_content).unwrap(); /// let mut model = des.into_model(); /// let (decision, _) = model.next_rwnd().unwrap(); -/// assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); +/// assert_eq!(decision.set_rcv_buf, Some(65536)); +/// assert_eq!(decision.app_read_bytes, Some(1024)); /// ``` pub struct RepeatedRwndPattern { pub pattern: Vec>, @@ -228,7 +230,7 @@ impl RwndTrace for StaticRwnd { if duration.is_zero() { None } else { - Some((self.decision.clone(), duration)) + Some((self.decision, duration)) } } else { None @@ -279,7 +281,7 @@ impl StaticRwndConfig { Self { duration: None, set_rcv_buf: None, - action: None, + app_read_bytes: None, } } @@ -288,18 +290,26 @@ impl StaticRwndConfig { self } + /// Size the receive buffer. + /// + /// On its own this also holds the advertised window at that value, with the + /// application draining continuously. Combined with [`Self::app_read`] it + /// only sets the size: the resize applies first, the read is taken against + /// the new size, and the window then follows `set_rcv_buf - unread` rather + /// than being pinned. pub fn set_rcv_buf(mut self, set_rcv_buf: u64) -> Self { self.set_rcv_buf = Some(set_rcv_buf); self } + /// The application reads exactly this many bytes over the step, then stops. + /// + /// The window follows from what is left unread of the standing buffer, or + /// of the buffer this step sets when combined with [`Self::set_rcv_buf`]. + /// The backlog carries across such a resize untouched, so the window + /// saturates at zero if the new size is at or below it. pub fn app_read(mut self, bytes: u64) -> Self { - self.action = Some(RwndAction::AppRead { bytes }); - self - } - - pub fn remaining(mut self, rwnd: u64) -> Self { - self.action = Some(RwndAction::Remaining { rwnd }); + self.app_read_bytes = Some(bytes); self } @@ -307,7 +317,7 @@ impl StaticRwndConfig { StaticRwnd { decision: RwndDecision { set_rcv_buf: self.set_rcv_buf, - action: self.action, + app_read_bytes: self.app_read_bytes, }, duration: Some(self.duration.unwrap_or_else(|| Duration::from_secs(1))), } @@ -360,11 +370,13 @@ impl_rwnd_trace_config!(RepeatedRwndPatternConfig); #[cfg(test)] mod test { use super::*; - use crate::model::StaticRwndConfig; use crate::RwndTrace; + /// Both fields on one step: the buffer is resized and the read is taken + /// against the new size, so the decision carries both rather than one + /// overriding the other. #[test] - fn test_static_rwnd_model_app_read() { + fn test_static_rwnd_model_buffer_and_app_read() { let mut static_rwnd = StaticRwndConfig::new() .set_rcv_buf(65536) .app_read(1024) @@ -372,35 +384,50 @@ mod test { .build(); let (decision, duration) = static_rwnd.next_rwnd().unwrap(); assert_eq!(decision.set_rcv_buf, Some(65536)); - assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); + assert_eq!(decision.app_read_bytes, Some(1024)); assert_eq!(duration, Duration::from_secs(1)); assert_eq!(static_rwnd.next_rwnd(), None); } + /// A buffer on its own is a complete statement: the window sits there and + /// the application keeps up. This is what used to need `rwnd_remaining`. #[test] - fn test_static_rwnd_model_remaining() { + fn test_static_rwnd_model_buffer_only() { let mut static_rwnd = StaticRwndConfig::new() - .remaining(32768) + .set_rcv_buf(32768) .duration(Duration::from_secs(2)) .build(); let (decision, duration) = static_rwnd.next_rwnd().unwrap(); - assert_eq!(decision.set_rcv_buf, None); - assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); + assert_eq!(decision.set_rcv_buf, Some(32768)); + assert_eq!(decision.app_read_bytes, None); assert_eq!(duration, Duration::from_secs(2)); assert_eq!(static_rwnd.next_rwnd(), None); } + /// A step may carry neither field, and then it simply holds whatever the + /// previous step configured for its duration. + #[test] + fn test_static_rwnd_model_carries_forward() { + let mut model = StaticRwndConfig::new() + .duration(Duration::from_secs(1)) + .build(); + let (decision, duration) = model.next_rwnd().unwrap(); + assert_eq!(decision.set_rcv_buf, None); + assert_eq!(decision.app_read_bytes, None); + assert_eq!(duration, Duration::from_secs(1)); + } + #[test] fn test_repeated_rwnd_pattern() { let pat = vec![ Box::new( StaticRwndConfig::new() - .app_read(1024) + .set_rcv_buf(65536) .duration(Duration::from_secs(1)), ) as Box, Box::new( StaticRwndConfig::new() - .remaining(32768) + .app_read(1024) .duration(Duration::from_secs(1)), ) as Box, ]; @@ -409,134 +436,80 @@ mod test { .count(2) .build(); let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::AppRead { bytes: 1024 })); + assert_eq!(next.0.set_rcv_buf, Some(65536)); assert_eq!(next.1, Duration::from_secs(1)); - let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::Remaining { rwnd: 32768 })); - let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::AppRead { bytes: 1024 })); - let next = model.next_rwnd().unwrap(); - assert_eq!(next.0.action, Some(RwndAction::Remaining { rwnd: 32768 })); + assert_eq!(model.next_rwnd().unwrap().0.app_read_bytes, Some(1024)); + assert_eq!(model.next_rwnd().unwrap().0.set_rcv_buf, Some(65536)); + assert_eq!(model.next_rwnd().unwrap().0.app_read_bytes, Some(1024)); assert_eq!(model.next_rwnd(), None); } #[test] #[cfg(feature = "serde")] - fn test_serde_roundtrip_app_read() { + fn test_serde_roundtrip_buffer_only() { let cfg = Box::new( StaticRwndConfig::new() .set_rcv_buf(65536) - .app_read(1024) .duration(Duration::from_secs(1)), ) as Box; let ser_str = serde_json::to_string(&cfg).unwrap(); #[cfg(feature = "human")] - let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}"; + let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":65536}}"; #[cfg(not(feature = "human"))] - let expected = "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536,\"app_read_bytes\":1024}}"; + let expected = + "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":65536}}"; assert_eq!(ser_str, expected); let des: Box = serde_json::from_str(&ser_str).unwrap(); let mut model = des.into_model(); let (decision, duration) = model.next_rwnd().unwrap(); assert_eq!(decision.set_rcv_buf, Some(65536)); - assert_eq!(decision.action, Some(RwndAction::AppRead { bytes: 1024 })); + assert_eq!(decision.app_read_bytes, None); assert_eq!(duration, Duration::from_secs(1)); } + /// Both keys on one step is legal now, and round-trips. #[test] #[cfg(feature = "serde")] - fn test_serde_roundtrip_remaining() { - let cfg = Box::new( - StaticRwndConfig::new() - .remaining(32768) - .duration(Duration::from_secs(1)), - ) as Box; - let ser_str = serde_json::to_string(&cfg).unwrap(); - #[cfg(feature = "human")] - let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"rwnd_remaining\":32768}}"; - #[cfg(not(feature = "human"))] - let expected = "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"rwnd_remaining\":32768}}"; - assert_eq!(ser_str, expected); - - let des: Box = serde_json::from_str(&ser_str).unwrap(); + fn test_serde_roundtrip_both_fields() { + let json = "{\"StaticRwndConfig\":{\"set_rcv_buf\":131072,\"app_read_bytes\":4096}}"; + let des: Box = serde_json::from_str(json).unwrap(); let mut model = des.into_model(); let (decision, _) = model.next_rwnd().unwrap(); - assert_eq!(decision.action, Some(RwndAction::Remaining { rwnd: 32768 })); - } - - #[test] - #[cfg(feature = "serde")] - fn test_serde_rejects_both() { - // Omit duration to avoid the human/non-human format ambiguity; we're testing - // the action constraint, not duration parsing. - let json = "{\"StaticRwndConfig\":{\"app_read_bytes\":1024,\"rwnd_remaining\":32768}}"; - let result: Result, _> = serde_json::from_str(json); - let err = result - .err() - .expect("deserialization should have failed") - .to_string(); - assert!( - err.contains("cannot set both"), - "expected 'cannot set both' in error, got: {err}" - ); - } - - #[test] - fn test_static_rwnd_set_rcv_buf_only() { - let mut model = StaticRwndConfig::new() - .set_rcv_buf(131072) - .duration(Duration::from_secs(1)) - .build(); - let (decision, duration) = model.next_rwnd().unwrap(); assert_eq!(decision.set_rcv_buf, Some(131072)); - assert_eq!(decision.action, None); - assert_eq!(duration, Duration::from_secs(1)); - assert_eq!(model.next_rwnd(), None); + assert_eq!(decision.app_read_bytes, Some(4096)); } #[test] #[cfg(feature = "serde")] - fn test_serde_roundtrip_set_rcv_buf_only() { - let cfg = Box::new( - StaticRwndConfig::new() - .set_rcv_buf(131072) - .duration(Duration::from_secs(1)), - ) as Box; + fn test_serde_omits_absent_fields() { + let cfg = Box::new(StaticRwndConfig::new().app_read(0)) as Box; let ser_str = serde_json::to_string(&cfg).unwrap(); - #[cfg(feature = "human")] - let expected = "{\"StaticRwndConfig\":{\"duration\":\"1s\",\"set_rcv_buf\":131072}}"; - #[cfg(not(feature = "human"))] - let expected = - "{\"StaticRwndConfig\":{\"duration\":{\"secs\":1,\"nanos\":0},\"set_rcv_buf\":131072}}"; - assert_eq!(ser_str, expected); - - let des: Box = serde_json::from_str(&ser_str).unwrap(); - let mut model = des.into_model(); - let (decision, duration) = model.next_rwnd().unwrap(); - assert_eq!(decision.set_rcv_buf, Some(131072)); - assert_eq!(decision.action, None); - assert_eq!(duration, Duration::from_secs(1)); - assert_eq!(model.next_rwnd(), None); + assert!(!ser_str.contains("set_rcv_buf"), "got: {ser_str}"); + assert!(ser_str.contains("app_read_bytes"), "got: {ser_str}"); } + /// A trace written against the old schema names a field that no longer + /// exists. Rejecting it is the point: silently ignoring `rwnd_remaining` + /// would turn every window step into a no-op and replay a trace that says + /// nothing, which looks like a healthy run producing wrong numbers. #[test] #[cfg(feature = "serde")] - fn test_serde_action_none_when_neither_set() { - // A step with only set_rcv_buf and no action fields should deserialize to action: None. - let json = "{\"StaticRwndConfig\":{\"set_rcv_buf\":65536}}"; - let des: Box = serde_json::from_str(json).unwrap(); - let mut model = des.into_model(); - let (decision, _) = model.next_rwnd().unwrap(); - assert_eq!(decision.set_rcv_buf, Some(65536)); - assert_eq!(decision.action, None); + fn test_serde_rejects_the_old_rwnd_remaining_field() { + let json = "{\"StaticRwndConfig\":{\"rwnd_remaining\":32768}}"; + let result: Result, _> = serde_json::from_str(json); + let err = result + .err() + .expect("a trace using the removed field should not deserialize") + .to_string(); + assert!(err.contains("rwnd_remaining"), "got: {err}"); } #[test] fn test_repeated_rwnd_pattern_all_zero_duration_terminates() { // All inner models have duration == 0 and return None immediately. - // With count == 0 (infinite repeat) the old recursive implementation - // would spin forever; the loop-based one must return None promptly. + // With count == 0 (infinite repeat) a recursive implementation would + // spin forever; the loop-based one must return None promptly. let pat = vec![ Box::new( StaticRwndConfig::new() @@ -545,7 +518,7 @@ mod test { ) as Box, Box::new( StaticRwndConfig::new() - .remaining(32768) + .set_rcv_buf(32768) .duration(Duration::ZERO), ) as Box, ];