use std::cell::RefCell;
use std::rc::Rc;
use libfw_core::{Capabilities, IntRange};
use serde::{Deserialize, Serialize};
use wasm_bindgen::JsValue;
pub const TUNE_WINDOW_MS: f64 = 1_000.0;
pub const GAIN_THRESHOLD: f64 = 0.05;
pub const RTT_INFLATE_THRESHOLD: f64 = 0.30;
pub const DEGRADE_FACTOR: f64 = 0.5;
pub const RTT_EWMA_ALPHA: f64 = 0.25;
pub const HARD_MAX_CONCURRENCY: i64 = 64;
pub const HARD_MAX_WINDOW: i64 = 64;
pub const HARD_MAX_CHUNK_SIZE: i64 = 16 * 1024 * 1024;
pub const HARD_MAX_IN_FLIGHT_BYTES: u64 = 256 * 1024 * 1024;
pub const HARD_MAX_IN_FLIGHT_REQUESTS: usize = 64;
pub const LEVEL_SAMPLE_SIZE: usize = 64 * 1024;
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "lowercase")]
pub enum TunePhase {
Uninitialized,
Ramping,
Settled,
Degraded,
}
#[derive(Debug, Clone, Copy, PartialEq, Default, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TuneStats {
pub rtt_ms: f64,
pub mbps: f64,
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TuneParams {
pub concurrency: usize,
pub upload_window: usize,
pub download_window: usize,
pub chunk_size: u64,
pub compress_level: i32,
}
pub fn request_budget(concurrency: usize, per_file_window: usize) -> usize {
concurrency
.max(1)
.saturating_mul(per_file_window.max(1))
.clamp(1, HARD_MAX_IN_FLIGHT_REQUESTS)
}
pub const DEFAULT_TUNE_TTL_MS: u64 = 60 * 60 * 1_000;
pub const TUNE_CACHE_VERSION: u32 = 1;
pub trait TuneStore {
fn get(&self, key: &str) -> Option<String>;
fn set(&self, key: &str, value: &str);
fn remove(&self, key: &str);
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TuneCache {
pub v: u32,
pub caps_hash: String,
pub saved_at_ms: u64,
pub params: TuneParamsCache,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub struct TuneParamsCache {
pub concurrency: usize,
pub upload_window: usize,
pub download_window: usize,
pub chunk_size: u64,
}
impl TuneParamsCache {
fn of(params: &TuneParams) -> Self {
TuneParamsCache {
concurrency: params.concurrency,
upload_window: params.upload_window,
download_window: params.download_window,
chunk_size: params.chunk_size,
}
}
fn into_params(self, compress_level: i32) -> TuneParams {
TuneParams {
concurrency: self.concurrency,
upload_window: self.upload_window,
download_window: self.download_window,
chunk_size: self.chunk_size,
compress_level,
}
}
}
pub fn tune_key(origin: &str, direction: TransferKind) -> String {
let dir = match direction {
TransferKind::Upload => "upload",
TransferKind::Download => "download",
};
format!("libfw.tune.v{TUNE_CACHE_VERSION}.{dir}.{origin}")
}
pub fn origin_of(url: &str) -> String {
let (scheme, rest) = match url.split_once("://") {
Some((s, r)) => (s, r),
None => return String::new(),
};
if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
return String::new();
}
let authority = rest
.split(['/', '?', '#'])
.next()
.unwrap_or("")
.trim();
if authority.is_empty() {
return String::new();
}
format!("{}://{}", scheme.to_ascii_lowercase(), authority.to_ascii_lowercase())
}
impl TuneParams {
pub fn request_budget(&self) -> usize {
request_budget(self.concurrency, self.upload_window.max(self.download_window))
}
pub fn from_caps_mins(caps: &Capabilities) -> TuneParams {
TuneParams {
concurrency: floor_of(&caps.limits.concurrency, HARD_MAX_CONCURRENCY) as usize,
upload_window: floor_of(&caps.limits.upload_window, HARD_MAX_WINDOW) as usize,
download_window: floor_of(&caps.limits.download_window, HARD_MAX_WINDOW) as usize,
chunk_size: floor_of(&caps.limits.chunk_size, HARD_MAX_CHUNK_SIZE) as u64,
compress_level: caps.compression.zrip_levels.min,
}
}
pub fn from_config(
concurrency: usize,
upload_window: usize,
download_window: usize,
chunk_size: u64,
compress_level: i32,
caps: &Capabilities,
) -> TuneParams {
TuneParams {
concurrency: bounded(
concurrency.max(1) as i64,
&caps.limits.concurrency,
HARD_MAX_CONCURRENCY,
) as usize,
upload_window: bounded(
upload_window.max(1) as i64,
&caps.limits.upload_window,
HARD_MAX_WINDOW,
) as usize,
download_window: bounded(
download_window.max(1) as i64,
&caps.limits.download_window,
HARD_MAX_WINDOW,
) as usize,
chunk_size: bounded(
chunk_size.max(1) as i64,
&caps.limits.chunk_size,
HARD_MAX_CHUNK_SIZE,
) as u64,
compress_level: caps.clamp_level(compress_level),
}
}
pub fn clamped_into(&self, caps: &Capabilities) -> TuneParams {
TuneParams {
concurrency: bounded(
self.concurrency as i64,
&caps.limits.concurrency,
HARD_MAX_CONCURRENCY,
) as usize,
upload_window: bounded(
self.upload_window as i64,
&caps.limits.upload_window,
HARD_MAX_WINDOW,
) as usize,
download_window: bounded(
self.download_window as i64,
&caps.limits.download_window,
HARD_MAX_WINDOW,
) as usize,
chunk_size: bounded(
self.chunk_size as i64,
&caps.limits.chunk_size,
HARD_MAX_CHUNK_SIZE,
) as u64,
compress_level: caps.clamp_level(self.compress_level),
}
}
pub fn upload_in_flight(&self) -> u64 {
self.concurrency as u64 * self.upload_window as u64 * self.chunk_size
}
pub fn download_in_flight(&self) -> u64 {
self.concurrency as u64 * self.download_window as u64 * self.chunk_size
}
}
fn floor_of(range: &IntRange, hard_max: i64) -> i64 {
range.min.clamp(1, hard_max.max(1))
}
fn bounded(v: i64, range: &IntRange, hard_max: i64) -> i64 {
let hard_max = hard_max.max(1);
let v = v.max(1);
if range.min > range.max {
return v.min(hard_max).max(1);
}
let lo = range.min.clamp(1, hard_max);
let hi = range.max.min(hard_max).max(lo);
v.clamp(lo, hi)
}
pub fn ewma(prev: Option<f64>, sample: f64, alpha: f64) -> f64 {
match prev {
None => sample,
Some(p) => alpha * sample + (1.0 - alpha) * p,
}
}
pub fn bdp_bytes(mbps: f64, rtt_ms: f64) -> u64 {
(mbps * 1e6 / 8.0 * rtt_ms / 1_000.0).max(0.0) as u64
}
pub const TARGET_CHUNK_SPAN_MS: f64 = 100.0;
const CHUNK_ALIGN: u64 = 64 * 1024;
pub fn chunk_for_link(mbps: f64, range: &IntRange, budget_max: u64) -> u64 {
let floor = floor_of(range, HARD_MAX_CHUNK_SIZE).max(1) as u64;
if !mbps.is_finite() || mbps <= 0.0 {
return floor.min(budget_max.max(1)).max(1);
}
let bytes = (mbps * 1e6 / 8.0 * TARGET_CHUNK_SPAN_MS / 1_000.0).clamp(0.0, i64::MAX as f64) as u64;
let aligned = (bytes / CHUNK_ALIGN).saturating_mul(CHUNK_ALIGN);
let clamped = bounded(aligned.max(1) as i64, range, HARD_MAX_CHUNK_SIZE) as u64;
clamped.clamp(1, budget_max.max(1))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RampDim {
Window,
Concurrency,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RampAction {
Raise,
AdvanceDim,
Settle,
Degrade,
Hold,
}
#[derive(Debug, Clone, Copy)]
pub struct RampInput {
pub mbps: f64,
pub prev_mbps: f64,
pub rtt_inflation: f64,
pub errors: u32,
pub degraded: bool,
pub ramping: bool,
pub low_gain_windows: u32,
pub in_flight: u64,
pub bdp: u64,
pub at_cap: bool,
pub last_dim: bool,
}
pub fn ramp_action(input: &RampInput) -> (RampAction, u32) {
if input.errors > 0 {
return (RampAction::Degrade, 0);
}
if input.at_cap {
return if input.last_dim {
(RampAction::Settle, 0)
} else {
(RampAction::AdvanceDim, 0)
};
}
if input.bdp > 0 && input.in_flight >= input.bdp {
return (RampAction::Settle, 0);
}
if input.rtt_inflation > RTT_INFLATE_THRESHOLD {
return (RampAction::Settle, 0);
}
let gain = if input.degraded {
0.0
} else if input.prev_mbps > 0.0 {
input.mbps / input.prev_mbps - 1.0
} else if input.ramping {
1.0
} else {
0.0
};
if input.ramping && gain >= GAIN_THRESHOLD {
return (RampAction::Raise, 0);
}
let low = input.low_gain_windows + 1;
if low >= 2 {
if input.degraded || input.last_dim {
(RampAction::Settle, low)
} else {
(RampAction::AdvanceDim, 0)
}
} else {
(RampAction::Hold, low)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompressLevel {
Auto,
Fast,
Balanced,
Max,
Fixed(i32),
}
pub fn resolve_level(level: CompressLevel, caps: &Capabilities) -> i32 {
let z = caps.compression.zrip_levels;
match level {
CompressLevel::Fast => z.min,
CompressLevel::Balanced | CompressLevel::Auto => z.default,
CompressLevel::Max => z.max,
CompressLevel::Fixed(l) => z.clamp_level(l),
}
}
pub fn choose_auto_level(
candidates: &[(i32, usize, f64)],
uncompressed: usize,
mbps: f64,
) -> i32 {
if candidates.is_empty() {
return 0;
}
let bytes_per_ms = if mbps > 0.0 {
mbps * 1e6 / 8.0 / 1_000.0
} else {
10e6 / 8.0 / 1_000.0
};
let max_saved = candidates
.iter()
.map(|&(_, clen, _)| uncompressed.saturating_sub(clen))
.max()
.unwrap_or(0);
if max_saved == 0 {
return candidates.iter().map(|&(l, _, _)| l).min().unwrap_or(0);
}
let mut best = (candidates[0].0, f64::NEG_INFINITY);
for &(lvl, clen, cms) in candidates {
let saved = uncompressed.saturating_sub(clen) as f64;
let score = saved / bytes_per_ms - cms;
if score > best.1 {
best = (lvl, score);
}
}
best.0
}
#[derive(Debug, Clone)]
pub struct TuneEvent {
pub phase: TunePhase,
pub params: TuneParams,
pub stats: TuneStats,
}
pub type TuneHandle = Rc<RefCell<TuningEngine>>;
pub struct TuningEngine {
enabled: bool,
level_cfg: CompressLevel,
policy_level: i32,
phase: TunePhase,
caps: Option<Capabilities>,
caps_hash: String,
params: TuneParams,
stats: TuneStats,
auto_level: Option<i32>,
auto_level_benchmarked: bool,
window_start_ms: f64,
window_done_base: u64,
window_errors: u32,
windows_closed: u32,
rtt_ewma: Option<f64>,
rtt_baseline: Option<f64>,
prev_mbps: f64,
low_gain_windows: u32,
dim: RampDim,
degraded_windows: u32,
settled: Option<TuneParams>,
settled_caps_hash: String,
settled_direction: Option<TransferKind>,
post_degrade_settle: bool,
direction: TransferKind,
store: Option<Rc<dyn TuneStore>>,
cache_ttl_ms: u64,
origin: String,
on_tuning: Option<js_sys::Function>,
}
impl TuningEngine {
pub fn new(enabled: bool, level_cfg: CompressLevel) -> TuningEngine {
TuningEngine {
enabled,
level_cfg,
policy_level: libfw_core::ZRIP_DEFAULT_LEVEL,
phase: TunePhase::Uninitialized,
caps: None,
caps_hash: String::new(),
params: TuneParams {
concurrency: 1,
upload_window: 1,
download_window: 1,
chunk_size: 1,
compress_level: 0,
},
stats: TuneStats::default(),
auto_level: None,
auto_level_benchmarked: false,
window_start_ms: 0.0,
window_done_base: 0,
window_errors: 0,
windows_closed: 0,
rtt_ewma: None,
rtt_baseline: None,
prev_mbps: 0.0,
low_gain_windows: 0,
dim: RampDim::Window,
degraded_windows: 0,
settled: None,
settled_caps_hash: String::new(),
settled_direction: None,
post_degrade_settle: false,
direction: TransferKind::Download,
store: None,
cache_ttl_ms: DEFAULT_TUNE_TTL_MS,
origin: String::new(),
on_tuning: None,
}
}
pub fn set_cache(&mut self, store: Rc<dyn TuneStore>, ttl_ms: u64) {
self.store = Some(store);
self.cache_ttl_ms = ttl_ms;
}
pub fn cache_ttl_ms(&self) -> u64 {
self.cache_ttl_ms
}
pub fn set_origin(&mut self, url: &str) {
self.origin = origin_of(url);
}
fn cache_key(&self) -> Option<String> {
if self.store.is_none() || self.cache_ttl_ms == 0 || self.origin.is_empty() {
return None;
}
Some(tune_key(&self.origin, self.direction))
}
fn remove_cache_entry(&self) {
if let (Some(store), Some(key)) = (&self.store, self.cache_key()) {
store.remove(&key);
}
}
fn load_cached_settle(&mut self, now_ms: f64, caps: &Capabilities) -> Option<TuneParams> {
let key = self.cache_key()?;
let store = self.store.clone()?;
let raw = store.get(&key)?;
let entry: TuneCache = match serde_json::from_str(&raw) {
Ok(entry) => entry,
Err(_) => {
store.remove(&key);
return None;
}
};
let aged = now_ms - entry.saved_at_ms as f64;
let fresh = aged >= 0.0 && aged <= self.cache_ttl_ms as f64;
if entry.v != TUNE_CACHE_VERSION || entry.caps_hash != caps.caps_hash() || !fresh {
store.remove(&key);
return None;
}
Some(entry.params.into_params(self.policy_level))
}
fn save_cached_settle(&self, now_ms: f64) {
let Some(key) = self.cache_key() else { return };
let Some(store) = self.store.as_ref() else { return };
let entry = TuneCache {
v: TUNE_CACHE_VERSION,
caps_hash: self.caps_hash.clone(),
saved_at_ms: if now_ms > 0.0 { now_ms as u64 } else { 0 },
params: TuneParamsCache::of(&self.params),
};
if let Ok(json) = serde_json::to_string(&entry) {
store.set(&key, &json);
}
}
pub fn set_direction(&mut self, kind: TransferKind) {
self.direction = kind;
}
pub fn direction(&self) -> TransferKind {
self.direction
}
pub fn set_on_tuning(&mut self, cb: Option<js_sys::Function>) {
self.on_tuning = cb;
}
pub fn phase(&self) -> TunePhase {
self.phase
}
pub fn params(&self) -> TuneParams {
self.params.clone()
}
pub fn stats(&self) -> TuneStats {
self.stats
}
pub fn enabled(&self) -> bool {
self.enabled
}
pub fn set_enabled(&mut self, enabled: bool) {
self.enabled = enabled;
}
pub fn caps(&self) -> Option<Capabilities> {
self.caps.clone()
}
pub fn state_js(&self) -> JsValue {
let o = js_sys::Object::new();
let phase = match self.phase {
TunePhase::Uninitialized => "uninitialized",
TunePhase::Ramping => "ramping",
TunePhase::Settled => "settled",
TunePhase::Degraded => "degraded",
};
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("phase"), &JsValue::from_str(phase));
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("params"), ¶ms_to_js(&self.params));
let stats = js_sys::Object::new();
let _ = js_sys::Reflect::set(
&stats,
&JsValue::from_str("rttMs"),
&JsValue::from_f64(self.stats.rtt_ms),
);
let _ = js_sys::Reflect::set(
&stats,
&JsValue::from_str("mbps"),
&JsValue::from_f64(self.stats.mbps),
);
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("stats"), &stats.into());
let _ = js_sys::Reflect::set(
&o,
&JsValue::from_str("capsHash"),
&JsValue::from_str(&self.caps_hash),
);
o.into()
}
pub fn caps_hash(&self) -> &str {
&self.caps_hash
}
pub fn begin_transfer(
&mut self,
caps: &Capabilities,
now_ms: f64,
static_params: &TuneParams,
) -> TuneParams {
self.caps = Some(caps.clone());
self.caps_hash = caps.caps_hash();
self.windows_closed = 0;
self.window_errors = 0;
self.window_done_base = 0;
self.window_start_ms = now_ms;
self.auto_level_benchmarked = false;
self.auto_level = None;
self.post_degrade_settle = false;
self.prev_mbps = 0.0;
self.low_gain_windows = 0;
self.rtt_ewma = None;
self.rtt_baseline = None;
self.degraded_windows = 0;
let policy_level = static_params.compress_level;
self.policy_level = policy_level;
if !self.enabled {
self.phase = TunePhase::Uninitialized;
self.params = static_params.clamped_into(caps);
self.enforce_in_flight_budget();
return self.params.clone();
}
let in_memory = self.settled.clone().filter(|_| {
self.settled_caps_hash == self.caps_hash
&& self.settled_direction == Some(self.direction)
});
let reusable = match in_memory {
Some(settled) => Some(settled),
None => self.load_cached_settle(now_ms, caps),
};
match reusable {
Some(settled) => {
self.params = settled.clamped_into(caps);
self.params.compress_level = policy_level;
self.phase = TunePhase::Settled;
}
None => {
self.params = TuneParams::from_caps_mins(caps);
self.params.compress_level = policy_level;
self.stats = TuneStats::default();
self.phase = TunePhase::Ramping;
}
}
self.dim = RampDim::Window;
self.enforce_in_flight_budget();
self.params.clone()
}
pub fn tick(
&mut self,
now_ms: f64,
done_bytes: u64,
rtt_ms: Option<f64>,
error: bool,
) -> Option<TuneEvent> {
if error {
self.window_errors += 1;
}
if let Some(r) = rtt_ms.filter(|&r| r > 0.0) {
let new_ewma = ewma(self.rtt_ewma, r, RTT_EWMA_ALPHA);
if self.enabled && self.rtt_baseline.is_none() {
self.rtt_baseline = Some(new_ewma);
}
self.rtt_ewma = Some(new_ewma);
}
let elapsed = now_ms - self.window_start_ms;
if elapsed < TUNE_WINDOW_MS {
return None;
}
if !self.enabled {
return None;
}
let caps = self.caps.clone()?;
let bytes = done_bytes.saturating_sub(self.window_done_base);
let mbps = if elapsed > 0.0 {
bytes as f64 / elapsed * 1_000.0 * 8.0 / 1e6
} else {
0.0
};
self.stats.mbps = mbps;
if let Some(rtt) = self.rtt_ewma {
self.stats.rtt_ms = rtt;
}
self.retune_chunk_size(mbps, &caps);
let ramping = self.phase == TunePhase::Ramping;
let input = RampInput {
mbps,
prev_mbps: self.prev_mbps,
rtt_inflation: if ramping { self.rtt_inflation() } else { 0.0 },
errors: self.window_errors,
degraded: self.phase == TunePhase::Degraded,
ramping,
low_gain_windows: self.low_gain_windows,
in_flight: self.in_flight(),
bdp: if ramping && self.dim == RampDim::Concurrency {
bdp_bytes(mbps, self.rtt_ewma.unwrap_or(self.stats.rtt_ms))
} else {
0
},
at_cap: self.dim_at_cap(&caps),
last_dim: self.dim == RampDim::Concurrency,
};
let (action, low) = ramp_action(&input);
self.low_gain_windows = low;
self.windows_closed += 1;
self.prev_mbps = mbps;
match action {
RampAction::Raise => {
self.raise_dim(&caps);
self.enforce_in_flight_budget();
if self.phase == TunePhase::Degraded {
self.phase = TunePhase::Ramping;
}
}
RampAction::AdvanceDim => {
self.dim = next_dim(self.dim);
}
RampAction::Settle => {
self.phase = TunePhase::Settled;
self.low_gain_windows = 0;
if !self.post_degrade_settle {
self.settled = Some(self.params.clone());
self.settled_caps_hash = self.caps_hash.clone();
self.settled_direction = Some(self.direction);
self.save_cached_settle(now_ms);
}
}
RampAction::Degrade => {
self.halve_dim();
self.phase = TunePhase::Degraded;
self.degraded_windows = 0;
self.low_gain_windows = 0;
}
RampAction::Hold => {
if self.phase == TunePhase::Degraded {
self.degraded_windows += 1;
if self.degraded_windows >= 2 {
self.phase = TunePhase::Settled;
self.post_degrade_settle = true;
}
}
}
}
self.window_start_ms = now_ms;
self.window_done_base = done_bytes;
self.window_errors = 0;
let event = TuneEvent {
phase: self.phase,
params: self.params.clone(),
stats: self.stats,
};
self.emit(&event);
Some(event)
}
pub fn transfer_end(&mut self, ok: bool) {
if !self.enabled {
return;
}
if !ok {
self.settled = None;
self.settled_caps_hash.clear();
self.settled_direction = None;
self.remove_cache_entry();
self.phase = TunePhase::Ramping;
}
}
pub fn upload_compress_level(
&mut self,
caps: &Capabilities,
sample: Option<&[u8]>,
mbps: f64,
) -> i32 {
match self.level_cfg {
CompressLevel::Auto => {
if !self.auto_level_benchmarked {
self.auto_level_benchmarked = true;
self.auto_level = Some(match sample {
Some(s) if !s.is_empty() => {
let s = &s[..LEVEL_SAMPLE_SIZE.min(s.len())];
let candidates = auto_candidates(caps);
let results = benchmark_levels(s, &candidates);
choose_auto_level(&results, s.len(), mbps.max(1.0))
}
_ => caps.compression.zrip_levels.default,
});
}
self.auto_level.unwrap_or(caps.compression.zrip_levels.default)
}
other => resolve_level(other, caps),
}
}
pub fn auto_level_ready(&self) -> bool {
self.auto_level_benchmarked
}
fn rtt_inflation(&self) -> f64 {
match (self.rtt_ewma, self.rtt_baseline) {
(Some(cur), Some(base)) if base > 0.0 => (cur / base - 1.0).max(0.0),
_ => 0.0,
}
}
fn in_flight(&self) -> u64 {
match self.last_transfer_kind() {
TransferKind::Upload => self.params.upload_in_flight(),
TransferKind::Download => self.params.download_in_flight(),
}
}
fn dim_value(&self) -> i64 {
match self.dim {
RampDim::Window => match self.last_transfer_kind() {
TransferKind::Upload => self.params.upload_window as i64,
TransferKind::Download => self.params.download_window as i64,
},
RampDim::Concurrency => self.params.concurrency as i64,
}
}
fn dim_max(&self, caps: &Capabilities) -> i64 {
let advertised = match self.dim {
RampDim::Window => match self.last_transfer_kind() {
TransferKind::Upload => caps.limits.upload_window.max,
TransferKind::Download => caps.limits.download_window.max,
},
RampDim::Concurrency => caps.limits.concurrency.max,
};
let hard = match self.dim {
RampDim::Window => HARD_MAX_WINDOW,
RampDim::Concurrency => HARD_MAX_CONCURRENCY,
};
advertised.min(hard).max(1)
}
fn max_chunk_for_budget(&self) -> u64 {
let widest = self.params.upload_window.max(self.params.download_window).max(1) as u64;
let divisor = (self.params.concurrency.max(1) as u64) * widest;
(HARD_MAX_IN_FLIGHT_BYTES / divisor).max(1)
}
fn retune_chunk_size(&mut self, mbps: f64, caps: &Capabilities) {
let target = chunk_for_link(mbps, &caps.limits.chunk_size, self.max_chunk_for_budget());
let current = self.params.chunk_size.max(1);
let grow = current.saturating_mul(3) / 2;
let shrink = (current / 2).max(1);
if target > grow || target < shrink {
self.params.chunk_size = target;
self.enforce_in_flight_budget();
}
}
fn enforce_in_flight_budget(&mut self) {
let max_chunk = self.max_chunk_for_budget();
self.params.chunk_size = self.params.chunk_size.clamp(1, max_chunk);
}
fn dim_at_cap(&self, caps: &Capabilities) -> bool {
self.dim_value() >= self.dim_max(caps)
}
fn raise_dim(&mut self, caps: &Capabilities) {
let dim_max: i64 = self.dim_max(caps);
let step = |v: i64| -> i64 {
if v <= 0 {
1
} else if v * 2 <= dim_max {
v * 2
} else {
(v + 1).min(dim_max)
}
};
match self.dim {
RampDim::Window => match self.last_transfer_kind() {
TransferKind::Upload => {
self.params.upload_window = step(self.params.upload_window as i64).max(1) as usize
}
TransferKind::Download => {
self.params.download_window =
step(self.params.download_window as i64).max(1) as usize
}
},
RampDim::Concurrency => {
self.params.concurrency = step(self.params.concurrency as i64).max(1) as usize
}
}
}
fn halve_dim(&mut self) {
let step = |v: i64| -> i64 { ((v as f64) * DEGRADE_FACTOR).floor().max(1.0) as i64 };
match self.dim {
RampDim::Window => match self.last_transfer_kind() {
TransferKind::Upload => {
self.params.upload_window = step(self.params.upload_window as i64) as usize
}
TransferKind::Download => {
self.params.download_window = step(self.params.download_window as i64) as usize
}
},
RampDim::Concurrency => {
self.params.concurrency = step(self.params.concurrency as i64) as usize
}
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TransferKind {
Upload,
Download,
}
impl TuningEngine {
fn last_transfer_kind(&self) -> TransferKind {
self.direction
}
}
fn next_dim(dim: RampDim) -> RampDim {
match dim {
RampDim::Window => RampDim::Concurrency,
RampDim::Concurrency => RampDim::Concurrency,
}
}
fn auto_candidates(caps: &Capabilities) -> Vec<i32> {
let z = caps.compression.zrip_levels;
vec![z.min, z.default, z.max]
}
fn benchmark_levels(sample: &[u8], levels: &[i32]) -> Vec<(i32, usize, f64)> {
levels
.iter()
.map(|&level| {
let t0 = now_ms();
let out = zrip::compress(sample, level).unwrap_or_else(|_| sample.to_vec());
(level, out.len(), now_ms() - t0)
})
.collect()
}
pub fn now_ms() -> f64 {
#[cfg(target_arch = "wasm32")]
{
js_sys::Date::now()
}
#[cfg(not(target_arch = "wasm32"))]
{
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map(|d| d.as_secs_f64() * 1_000.0)
.unwrap_or(0.0)
}
}
impl TuningEngine {
fn emit(&self, event: &TuneEvent) {
if let Some(cb) = &self.on_tuning {
let params = params_to_js(&event.params);
let stats = js_sys::Object::new();
let _ = js_sys::Reflect::set(
&stats,
&JsValue::from_str("rttMs"),
&JsValue::from_f64(event.stats.rtt_ms),
);
let _ = js_sys::Reflect::set(
&stats,
&JsValue::from_str("mbps"),
&JsValue::from_f64(event.stats.mbps),
);
let phase = match event.phase {
TunePhase::Uninitialized => "uninitialized",
TunePhase::Ramping => "ramping",
TunePhase::Settled => "settled",
TunePhase::Degraded => "degraded",
};
let _ = cb.call3(
&JsValue::NULL,
&JsValue::from_str(phase),
¶ms,
&stats.into(),
);
}
}
}
fn params_to_js(params: &TuneParams) -> JsValue {
let o = js_sys::Object::new();
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("concurrency"), &JsValue::from_f64(params.concurrency as f64));
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("uploadWindow"), &JsValue::from_f64(params.upload_window as f64));
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("downloadWindow"), &JsValue::from_f64(params.download_window as f64));
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("chunkSize"), &JsValue::from_f64(params.chunk_size as f64));
let _ = js_sys::Reflect::set(&o, &JsValue::from_str("compressLevel"), &JsValue::from_f64(params.compress_level as f64));
o.into()
}
#[cfg(test)]
mod tests {
use super::*;
fn caps() -> Capabilities {
Capabilities::default()
}
fn static_params() -> TuneParams {
TuneParams {
concurrency: 4,
upload_window: 8,
download_window: 4,
chunk_size: 2 * 1024 * 1024,
compress_level: 1,
}
}
fn engine() -> TuningEngine {
let mut e = TuningEngine::new(true, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
e.begin_transfer(&caps(), 0.0, &static_params());
e
}
fn ramp_to_settle(e: &mut TuningEngine) {
let step = 12_500_000u64;
let mut done = step;
for i in 0..16 {
let t = 1_000.0 * (i as f64 + 1.0);
if let Some(TuneEvent {
phase: TunePhase::Settled,
..
}) = e.tick(t, done, Some(400.0), false)
{
return;
}
done += step;
}
panic!("engine did not settle");
}
#[test]
fn ramp_action_error_degrades() {
let input = RampInput {
mbps: 10.0,
prev_mbps: 5.0,
rtt_inflation: 0.0,
errors: 1,
degraded: false,
ramping: true,
low_gain_windows: 0,
in_flight: 100,
bdp: 1_000,
at_cap: false,
last_dim: false,
};
assert_eq!(ramp_action(&input).0, RampAction::Degrade);
}
#[test]
fn ramp_action_cap_advances_or_settles() {
let base = RampInput {
mbps: 10.0,
prev_mbps: 5.0,
rtt_inflation: 0.0,
errors: 0,
degraded: false,
ramping: true,
low_gain_windows: 0,
in_flight: 100,
bdp: 1_000,
at_cap: true,
last_dim: false,
};
assert_eq!(ramp_action(&base).0, RampAction::AdvanceDim);
let last = RampInput { last_dim: true, ..base };
assert_eq!(ramp_action(&last).0, RampAction::Settle);
}
#[test]
fn ramp_action_bdp_and_rtt_saturate() {
let base = RampInput {
mbps: 10.0,
prev_mbps: 5.0,
rtt_inflation: 0.0,
errors: 0,
degraded: false,
ramping: true,
low_gain_windows: 0,
in_flight: 2_000,
bdp: 1_000,
at_cap: false,
last_dim: false,
};
assert_eq!(ramp_action(&base).0, RampAction::Settle, "in-flight ≥ BDP settles");
let rtt = RampInput {
rtt_inflation: 0.5,
in_flight: 10,
bdp: 1_000,
..base
};
assert_eq!(ramp_action(&rtt).0, RampAction::Settle, "RTT inflation settles");
}
#[test]
fn ramp_action_gain_raises_and_low_gain_settles() {
let growing = RampInput {
mbps: 20.0,
prev_mbps: 10.0,
..RampInput {
mbps: 0.0,
prev_mbps: 0.0,
rtt_inflation: 0.0,
errors: 0,
degraded: false,
ramping: true,
low_gain_windows: 7,
in_flight: 10,
bdp: 1_000,
at_cap: false,
last_dim: false,
}
};
let (action, low) = ramp_action(&growing);
assert_eq!(action, RampAction::Raise);
assert_eq!(low, 0, "raise resets the low-gain counter");
let flat = RampInput {
mbps: 10.4,
prev_mbps: 10.0,
low_gain_windows: 0,
..growing
};
let (action, low) = ramp_action(&flat);
assert_eq!(action, RampAction::Hold);
assert_eq!(low, 1, "first below-threshold window holds");
let flat2 = RampInput { low_gain_windows: 1, ..flat };
let (action, low) = ramp_action(&flat2);
assert_eq!(action, RampAction::AdvanceDim);
assert_eq!(low, 0, "advancing resets the low-gain counter");
let flat_last = RampInput {
last_dim: true,
low_gain_windows: 1,
..flat
};
assert_eq!(ramp_action(&flat_last).0, RampAction::Settle);
let first = RampInput {
mbps: 1.0,
prev_mbps: 0.0,
low_gain_windows: 0,
..growing
};
assert_eq!(ramp_action(&first).0, RampAction::Raise);
}
#[test]
fn ramp_action_degraded_hold_never_raises() {
let degraded = RampInput {
mbps: 0.008,
prev_mbps: 0.0,
degraded: true,
..RampInput {
mbps: 0.0,
prev_mbps: 0.0,
rtt_inflation: 0.0,
errors: 0,
degraded: false,
ramping: true,
low_gain_windows: 0,
in_flight: 10,
bdp: 1_000,
at_cap: false,
last_dim: false,
}
};
let (action, low) = ramp_action(°raded);
assert_eq!(action, RampAction::Hold);
assert_eq!(low, 1, "first stability window holds");
let (action, low) = ramp_action(&RampInput { low_gain_windows: 1, ..degraded });
assert_eq!(action, RampAction::Settle);
assert_eq!(low, 2, "second stability window settles");
}
#[test]
fn ewma_seeds_and_smooths() {
assert_eq!(ewma(None, 10.0, 0.25), 10.0);
let v = ewma(Some(10.0), 20.0, 0.25);
assert!((v - 12.5).abs() < 1e-9);
}
#[test]
fn bdp_conversion() {
assert_eq!(bdp_bytes(10.0, 100.0), 125_000);
assert_eq!(bdp_bytes(0.0, 0.0), 0);
}
#[test]
fn request_budget_covers_files_and_the_per_file_window() {
assert_eq!(request_budget(1, 1), 1);
assert_eq!(request_budget(1, 8), 8);
assert_eq!(request_budget(4, 8), 32);
assert_eq!(request_budget(0, 0), 1);
assert_eq!(request_budget(64, 64), HARD_MAX_IN_FLIGHT_REQUESTS);
let mut p = static_params();
p.concurrency = 2;
p.upload_window = 8;
p.download_window = 4;
assert_eq!(p.request_budget(), 16);
p.upload_window = 4;
assert_eq!(p.request_budget(), 8);
}
#[test]
fn resolve_level_mapping() {
let c = caps();
assert_eq!(resolve_level(CompressLevel::Fast, &c), -8);
assert_eq!(resolve_level(CompressLevel::Balanced, &c), 1);
assert_eq!(resolve_level(CompressLevel::Max, &c), 4);
assert_eq!(resolve_level(CompressLevel::Auto, &c), 1);
assert_eq!(resolve_level(CompressLevel::Fixed(2), &c), 2);
assert_eq!(resolve_level(CompressLevel::Fixed(99), &c), 4);
assert_eq!(resolve_level(CompressLevel::Fixed(-99), &c), -8);
}
#[test]
fn choose_auto_level_picks_by_net_gain() {
assert_eq!(
choose_auto_level(&[(-8, 1000, 0.1), (1, 1000, 5.0), (4, 1000, 40.0)], 1000, 10.0),
-8
);
let cpu_bound = choose_auto_level(&[(-8, 800_000, 0.1), (4, 400_000, 30.0)], 1_000_000, 1000.0);
assert_eq!(cpu_bound, -8);
let savings = choose_auto_level(&[(-8, 800_000, 0.1), (4, 400_000, 30.0)], 1_000_000, 10.0);
assert_eq!(savings, 4);
assert_eq!(choose_auto_level(&[(0, 500, 1.0), (4, 500, 1.0)], 1000, 100.0), 0);
}
#[test]
fn begin_transfer_miss_ramps_from_mins() {
let mut e = TuningEngine::new(true, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
let p = e.begin_transfer(&caps(), 0.0, &static_params());
assert_eq!(e.phase(), TunePhase::Ramping);
assert_eq!(p.concurrency, 1);
assert_eq!(p.upload_window, 1);
assert_eq!(p.chunk_size, 262_144);
}
#[test]
fn settled_reuse_first_window_does_not_raise() {
let input = RampInput {
mbps: 100.0,
prev_mbps: 0.0,
rtt_inflation: 0.0,
errors: 0,
degraded: false,
ramping: false,
low_gain_windows: 0,
in_flight: 10,
bdp: 0,
at_cap: false,
last_dim: false,
};
let (action, low) = ramp_action(&input);
assert_eq!(action, RampAction::Hold);
assert_eq!(low, 1);
}
#[test]
fn reused_params_do_not_raise_on_the_first_window() {
let mut e = engine();
ramp_to_settle(&mut e);
let tuned = e.params().clone();
e.transfer_end(true);
e.begin_transfer(&caps(), 60_000.0, &static_params());
let ev = e.tick(61_000.0, 100_000_000, Some(400.0), false).unwrap();
assert_eq!(ev.phase, TunePhase::Settled);
assert_eq!(
ev.params.upload_window, tuned.upload_window,
"settled reuse must not raise the window"
);
assert_eq!(ev.params.concurrency, tuned.concurrency);
}
#[test]
fn rtt_inflation_settles_the_ramp() {
let mut e = engine();
let ev = e.tick(1_000.0, 12_500_000, Some(100.0), false).unwrap();
assert_eq!(ev.phase, TunePhase::Ramping);
let ev = e.tick(2_000.0, 25_000_000, Some(400.0), false).unwrap();
assert_eq!(ev.phase, TunePhase::Settled, "RTT inflation must settle");
}
#[test]
fn ramp_start_keeps_the_configured_compress_policy() {
let mut e = TuningEngine::new(true, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
let sp = static_params(); let p = e.begin_transfer(&caps(), 0.0, &sp);
assert_eq!(
p.compress_level, 1,
"a ramp must not silently drop to the advertised minimum (-8)"
);
ramp_to_settle(&mut e);
e.transfer_end(true);
let p2 = e.begin_transfer(&caps(), 60_000.0, &sp);
assert_eq!(p2.compress_level, 1, "reuse keeps the policy level");
}
#[test]
fn absurd_advertised_caps_are_clamped_to_hard_ceilings() {
let mut hostile = caps();
hostile.limits.concurrency = IntRange {
default: 1,
max: 100_000,
min: 1_000,
};
hostile.limits.upload_window = IntRange {
default: 1,
max: 100_000,
min: 500,
};
hostile.limits.download_window = IntRange {
default: 1,
max: 100_000,
min: 500,
};
hostile.limits.chunk_size = IntRange {
default: i64::MAX,
max: i64::MAX,
min: i64::MAX,
};
let mut e = TuningEngine::new(true, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
let p = e.begin_transfer(&hostile, 0.0, &static_params());
assert!(p.concurrency <= HARD_MAX_CONCURRENCY as usize);
assert!(p.upload_window <= HARD_MAX_WINDOW as usize);
assert!(p.download_window <= HARD_MAX_WINDOW as usize);
assert!(p.chunk_size <= HARD_MAX_CHUNK_SIZE as u64);
assert!(
p.upload_in_flight() <= HARD_MAX_IN_FLIGHT_BYTES,
"in-flight bytes {} exceed the budget",
p.upload_in_flight()
);
}
#[test]
fn inverted_advertised_range_does_not_panic() {
let mut broken = caps();
broken.limits.concurrency = IntRange {
default: 4,
max: 2,
min: 8,
};
broken.limits.chunk_size = IntRange {
default: 0,
max: 0,
min: 0,
};
let mut e = TuningEngine::new(false, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
let p = e.begin_transfer(&broken, 0.0, &static_params());
assert!(p.concurrency >= 1 && p.chunk_size >= 1);
let mut e2 = TuningEngine::new(true, CompressLevel::Balanced);
e2.set_direction(TransferKind::Upload);
let p2 = e2.begin_transfer(&broken, 0.0, &static_params());
assert!(p2.concurrency >= 1 && p2.chunk_size >= 1);
}
#[test]
fn ramp_raises_window_then_concurrency_until_cap() {
let mut e = engine();
let mut t = 1_000.0;
let mut done = 12_500_000u64;
let mut delta = 12_500_000u64;
for _ in 0..3 {
let ev = e.tick(t, done, Some(400.0), false).expect("window closed");
assert_eq!(ev.phase, TunePhase::Ramping);
t += 1_000.0;
delta *= 2;
done += delta; }
assert_eq!(e.params().upload_window, 8);
let ev = e.tick(t, done, Some(400.0), false).unwrap();
assert_eq!(ev.params.upload_window, 8); t += 1_000.0;
delta *= 2;
done += delta;
let ev = e.tick(t, done, Some(400.0), false).unwrap();
assert_eq!(ev.params.concurrency, 2);
let ev = e.tick(t + 1_000.0, done + delta * 2, Some(400.0), false).unwrap();
assert_eq!(ev.params.concurrency, 4);
assert_eq!(ev.params.upload_window, 8);
}
#[test]
fn error_window_degrades_and_halves() {
let mut e = engine();
let mut t = 1_000.0;
let mut done = 12_500_000u64;
let mut delta = 12_500_000u64;
for _ in 0..2 {
e.tick(t, done, Some(400.0), false);
t += 1_000.0;
delta *= 2;
done += delta; }
assert_eq!(e.params().upload_window, 4);
let ev = e.tick(t, done, Some(400.0), true).unwrap();
assert_eq!(ev.phase, TunePhase::Degraded);
assert_eq!(ev.params.upload_window, 2);
let ev = e.tick(t + 1_000.0, done + 1_000, None, false).unwrap();
assert_eq!(ev.phase, TunePhase::Degraded);
let ev = e.tick(t + 2_000.0, done + 2_000, None, false).unwrap();
assert_eq!(ev.phase, TunePhase::Settled);
assert_eq!(ev.params.upload_window, 2);
}
#[test]
fn flat_gain_explores_dimensions_then_settles() {
let mut e = engine();
let step = 12_500_000u64;
let mut done = step;
let mut t = 1_000.0;
let ev = e.tick(t, done, Some(400.0), false).unwrap();
assert_eq!(ev.params.upload_window, 2);
for _ in 0..2 {
t += 1_000.0;
done += step;
e.tick(t, done, Some(400.0), false);
}
assert_eq!(
e.phase(),
TunePhase::Ramping,
"a flat window dimension must not end the ramp"
);
t += 1_000.0;
done += step;
let ev = e.tick(t, done, Some(400.0), false).unwrap();
assert_eq!(ev.phase, TunePhase::Ramping);
t += 1_000.0;
done += step;
let ev = e.tick(t, done, Some(400.0), false).unwrap();
assert_eq!(ev.phase, TunePhase::Settled);
assert_eq!(ev.params.upload_window, 2);
e.transfer_end(true);
}
#[test]
fn chunk_size_tracks_the_link_and_ignores_jitter() {
let mut e = engine();
let mut done = 12_500_000u64;
let ev = e.tick(1_000.0, done, Some(400.0), false).unwrap();
let first = ev.params.chunk_size;
assert_eq!(first, 1_245_184, "the chunk must follow the measured link");
let mut t = 1_000.0;
for _ in 0..3 {
t += 1_000.0;
done += 13_500_000; let ev = e.tick(t, done, Some(400.0), false).unwrap();
assert_eq!(ev.params.chunk_size, first, "jitter must not re-size the chunk");
}
t += 1_000.0;
done += 100_000_000; let ev = e.tick(t, done, Some(400.0), false).unwrap();
assert!(ev.params.chunk_size > first, "the chunk must grow on a wider band");
}
#[test]
fn chunk_size_follows_the_measured_link() {
let range = IntRange {
min: 262_144,
default: 2 * 1024 * 1024,
max: 8 * 1024 * 1024,
};
let budget = 8 * 1024 * 1024;
assert_eq!(chunk_for_link(0.0, &range, budget), 262_144);
assert_eq!(chunk_for_link(10.0, &range, budget), 262_144);
assert_eq!(chunk_for_link(30.0, &range, budget), 327_680);
assert_eq!(chunk_for_link(100.0, &range, budget), 1_245_184);
assert_eq!(chunk_for_link(1_000.0, &range, budget), 8 * 1024 * 1024);
assert_eq!(chunk_for_link(1_000.0, &range, 1_500_000), 1_500_000);
let mut last = 0;
for mbps in [1.0, 5.0, 20.0, 80.0, 320.0] {
let v = chunk_for_link(mbps, &range, budget);
assert!(v >= last && v >= 262_144, "{mbps} Mb/s → {v}");
last = v;
}
assert!(chunk_for_link(f64::NAN, &range, budget) >= 1);
assert!(chunk_for_link(-5.0, &range, budget) >= 1);
let inverted = IntRange {
min: 9_000_000,
max: 1,
default: 0,
};
let v = chunk_for_link(50.0, &inverted, budget);
assert!(v >= 1 && v <= budget, "{v}");
}
#[test]
fn settled_state_is_reused_within_the_same_client() {
let mut e = engine();
ramp_to_settle(&mut e);
assert_eq!(e.phase(), TunePhase::Settled);
let tuned = e.params().clone();
e.transfer_end(true);
let p = e.begin_transfer(&caps(), 60_000.0, &static_params());
assert_eq!(e.phase(), TunePhase::Settled, "in-session reuse skips the ramp");
assert_eq!(p.upload_window, tuned.upload_window);
assert_eq!(p.concurrency, tuned.concurrency);
assert_eq!(p.chunk_size, tuned.chunk_size);
assert_eq!(p.compress_level, 1, "reuse keeps the configured policy level");
}
#[test]
fn a_new_engine_always_re_ramps_from_mins() {
let mut e = engine();
ramp_to_settle(&mut e);
e.transfer_end(true);
let fresh = engine();
assert_eq!(fresh.phase(), TunePhase::Ramping);
assert_eq!(fresh.params().upload_window, 1);
assert_eq!(fresh.params().chunk_size, 262_144);
}
#[derive(Clone, Default)]
struct FakeStore(Rc<RefCell<std::collections::HashMap<String, String>>>);
impl FakeStore {
fn len(&self) -> usize {
self.0.borrow().len()
}
fn put_raw(&self, key: &str, value: &str) {
self.0.borrow_mut().insert(key.to_string(), value.to_string());
}
fn keys(&self) -> Vec<String> {
self.0.borrow().keys().cloned().collect()
}
}
impl TuneStore for FakeStore {
fn get(&self, key: &str) -> Option<String> {
self.0.borrow().get(key).cloned()
}
fn set(&self, key: &str, value: &str) {
self.0.borrow_mut().insert(key.to_string(), value.to_string());
}
fn remove(&self, key: &str) {
self.0.borrow_mut().remove(key);
}
}
const ORIGIN: &str = "http://localhost:8080";
fn wired(
store: &FakeStore,
ttl_ms: u64,
direction: TransferKind,
origin: &str,
) -> TuningEngine {
let mut e = TuningEngine::new(true, CompressLevel::Balanced);
e.set_cache(Rc::new(store.clone()), ttl_ms);
e.set_direction(direction);
e.set_origin(origin);
e
}
fn cached_engine(store: &FakeStore, ttl_ms: u64) -> TuningEngine {
let mut e = wired(store, ttl_ms, TransferKind::Upload, ORIGIN);
e.begin_transfer(&caps(), 0.0, &static_params());
e
}
#[test]
fn a_settle_is_cached_and_reused_by_a_new_engine() {
let store = FakeStore::default();
let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
ramp_to_settle(&mut a);
assert_eq!(a.phase(), TunePhase::Settled);
let tuned = a.params().clone();
a.transfer_end(true);
assert_eq!(store.len(), 1, "the settle is written on convergence");
assert_eq!(
store.keys(),
vec![tune_key(ORIGIN, TransferKind::Upload)],
"keyed by origin + direction"
);
let mut b = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
let p = b.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(b.phase(), TunePhase::Settled, "a cache hit skips the ramp");
assert_eq!(p.upload_window, tuned.upload_window);
assert_eq!(p.concurrency, tuned.concurrency);
assert_eq!(p.chunk_size, tuned.chunk_size);
assert_eq!(p.compress_level, 1, "the policy level is not cached");
}
#[test]
fn a_cached_settle_expires_after_the_ttl() {
let store = FakeStore::default();
let ttl = 60_000;
let mut a = cached_engine(&store, ttl);
ramp_to_settle(&mut a);
let saved = a.params().clone();
assert!(saved.upload_window > 1, "the ramp must have raised something");
a.transfer_end(true);
let row: TuneCache = serde_json::from_str(
&store.get(&tune_key(ORIGIN, TransferKind::Upload)).unwrap(),
)
.unwrap();
let saved_at = row.saved_at_ms as f64;
let mut b = wired(&store, ttl, TransferKind::Upload, ORIGIN);
b.begin_transfer(&caps(), saved_at + ttl as f64 - 1.0, &static_params());
assert_eq!(b.phase(), TunePhase::Settled);
let mut c = wired(&store, ttl, TransferKind::Upload, ORIGIN);
let p = c.begin_transfer(&caps(), saved_at + ttl as f64 + 1.0, &static_params());
assert_eq!(c.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1, "an expired result re-ramps from the mins");
assert_eq!(store.len(), 0, "an expired row is cleaned up");
}
#[test]
fn a_zero_ttl_disables_the_cache() {
let store = FakeStore::default();
assert_eq!(cached_engine(&store, 0).cache_ttl_ms(), 0);
let mut a = cached_engine(&store, 0);
ramp_to_settle(&mut a);
a.transfer_end(true);
assert_eq!(store.len(), 0, "ttl = 0 never writes");
let store2 = FakeStore::default();
let mut x = cached_engine(&store2, DEFAULT_TUNE_TTL_MS);
ramp_to_settle(&mut x);
x.transfer_end(true);
assert_eq!(store2.len(), 1);
let mut y = wired(&store2, 0, TransferKind::Upload, ORIGIN);
y.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(y.phase(), TunePhase::Ramping, "ttl = 0 never reads");
}
#[test]
fn a_cached_settle_is_scoped_to_one_direction() {
let store = FakeStore::default();
let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
ramp_to_settle(&mut a);
a.transfer_end(true);
let mut b = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Download, ORIGIN);
let p = b.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(b.phase(), TunePhase::Ramping);
assert_eq!(p.download_window, 1);
let mut c = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
c.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(c.phase(), TunePhase::Settled);
}
#[test]
fn a_cached_settle_is_scoped_to_one_origin() {
let store = FakeStore::default();
let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
ramp_to_settle(&mut a);
a.transfer_end(true);
let mut b = wired(
&store,
DEFAULT_TUNE_TTL_MS,
TransferKind::Upload,
"https://other.example.com:8443",
);
b.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(b.phase(), TunePhase::Ramping, "another origin is another link");
assert_eq!(store.len(), 1, "the other origin's row is left alone");
}
#[test]
fn an_unusable_origin_disables_the_cache() {
let store = FakeStore::default();
for url in ["", "not a url", "ftp://host/x", "http://", "ws://h:80"] {
let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, url);
let _ = e.begin_transfer(&caps(), 0.0, &static_params());
ramp_to_settle(&mut e);
e.transfer_end(true);
assert_eq!(store.len(), 0, "{url:?} must not be cached");
}
}
#[test]
fn changed_caps_invalidate_a_cached_settle() {
let store = FakeStore::default();
let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
ramp_to_settle(&mut a);
a.transfer_end(true);
assert_eq!(store.len(), 1);
let mut changed = caps();
changed.limits.concurrency.default = 6;
let mut b = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
let p = b.begin_transfer(&changed, 20_000.0, &static_params());
assert_eq!(b.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1);
assert_eq!(store.len(), 0, "a caps mismatch drops the entry");
}
#[test]
fn a_failed_transfer_removes_the_cached_settle() {
let store = FakeStore::default();
let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
ramp_to_settle(&mut a);
a.transfer_end(false);
assert_eq!(store.len(), 0, "parameters that just failed are not kept");
let p = a.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(a.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1);
}
#[test]
fn a_corrupt_or_foreign_cache_entry_is_dropped() {
let store = FakeStore::default();
let key = tune_key(ORIGIN, TransferKind::Upload);
store.put_raw(&key, "{not json");
let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
e.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(e.phase(), TunePhase::Ramping);
assert_eq!(store.len(), 0, "an unreadable row is removed");
let future = TuneCache {
v: TUNE_CACHE_VERSION + 1,
caps_hash: caps().caps_hash(),
saved_at_ms: 1_000,
params: TuneParamsCache {
concurrency: 12,
upload_window: 16,
download_window: 16,
chunk_size: 4 * 1024 * 1024,
},
};
store.put_raw(&key, &serde_json::to_string(&future).unwrap());
let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
e.begin_transfer(&caps(), 20_000.0, &static_params());
assert_eq!(e.phase(), TunePhase::Ramping);
assert_eq!(store.len(), 0);
}
#[test]
fn a_valid_row_written_by_another_client_is_reused() {
let store = FakeStore::default();
let row = TuneCache {
v: TUNE_CACHE_VERSION,
caps_hash: caps().caps_hash(),
saved_at_ms: 5_000,
params: TuneParamsCache {
concurrency: 3,
upload_window: 7,
download_window: 3,
chunk_size: 1_048_576,
},
};
store.put_raw(
&tune_key(ORIGIN, TransferKind::Upload),
&serde_json::to_string(&row).unwrap(),
);
let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
let p = e.begin_transfer(&caps(), 10_000.0, &static_params());
assert_eq!(e.phase(), TunePhase::Settled);
assert_eq!(p.concurrency, 3);
assert_eq!(p.upload_window, 7);
assert_eq!(p.chunk_size, 1_048_576);
assert_eq!(store.len(), 1, "a fresh, valid row is kept");
}
#[test]
fn tune_key_and_origin_of_normalise_the_scope() {
assert_eq!(
tune_key("http://localhost:8080", TransferKind::Upload),
"libfw.tune.v1.upload.http://localhost:8080"
);
assert_eq!(
tune_key("http://localhost:8080", TransferKind::Download),
"libfw.tune.v1.download.http://localhost:8080"
);
assert_ne!(
tune_key("http://a", TransferKind::Upload),
tune_key("http://a", TransferKind::Download)
);
assert_eq!(origin_of("http://Host:8080/x?y#z"), "http://host:8080");
assert_eq!(origin_of("HTTPS://h/p"), "https://h");
assert_eq!(origin_of("http://h"), "http://h");
assert_eq!(origin_of(""), "");
assert_eq!(origin_of("/relative/path"), "");
assert_eq!(origin_of("ftp://h/x"), "");
assert_eq!(origin_of("http://"), "");
}
#[test]
fn failed_transfer_forgets_the_settled_state() {
let mut e = engine();
ramp_to_settle(&mut e);
e.transfer_end(false);
let p = e.begin_transfer(&caps(), 60_000.0, &static_params());
assert_eq!(e.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1, "a failed transfer re-ramps from the mins");
}
#[test]
fn changed_caps_invalidate_the_settled_state() {
let mut e = engine();
ramp_to_settle(&mut e);
e.transfer_end(true);
let mut changed = caps();
changed.limits.concurrency.default = 6;
let p = e.begin_transfer(&changed, 60_000.0, &static_params());
assert_eq!(e.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1);
}
#[test]
fn reused_settle_does_not_re_ramp_on_later_windows() {
let mut e = engine();
ramp_to_settle(&mut e);
let tuned = e.params().clone();
e.transfer_end(true);
e.begin_transfer(&caps(), 60_000.0, &static_params());
let mut t = 61_000.0;
for i in 1..=3u64 {
let ev = e.tick(t, i * 100_000_000, Some(400.0), false).unwrap();
assert_eq!(ev.phase, TunePhase::Settled);
assert_eq!(ev.params.upload_window, tuned.upload_window);
assert_eq!(ev.params.concurrency, tuned.concurrency);
t += 1_000.0;
}
}
#[test]
fn transfer_end_without_settle_keeps_ramping_next_time() {
let mut e = engine();
e.tick(1_000.0, 12_500_000, Some(400.0), false);
e.transfer_end(true);
let p = e.begin_transfer(&caps(), 60_000.0, &static_params());
assert_eq!(e.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1);
}
#[test]
fn upload_auto_level_benchmarks_once() {
let mut e = TuningEngine::new(true, CompressLevel::Auto);
e.begin_transfer(&caps(), 0.0, &static_params());
let sample = vec![0xABu8; 256 * 1024];
let lvl1 = e.upload_compress_level(&caps(), Some(&sample), 50.0);
let lvl2 = e.upload_compress_level(&caps(), Some(&sample), 50.0);
assert_eq!(lvl1, lvl2, "benchmark runs once, result cached");
assert!(e.auto_level_ready());
assert!((-8..=4).contains(&lvl1), "level inside advertised range, got {lvl1}");
let mut e2 = TuningEngine::new(true, CompressLevel::Auto);
e2.begin_transfer(&caps(), 0.0, &static_params());
let lvl = e2.upload_compress_level(&caps(), Some(&[0u8; 0]), 50.0);
assert_eq!(lvl, caps().compression.zrip_levels.default, "no sample → server default");
}
#[test]
fn disabled_engine_uses_static_params() {
let mut e = TuningEngine::new(false, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
let p = e.begin_transfer(&caps(), 0.0, &static_params());
assert_eq!(p.concurrency, 4, "static config preserved");
assert_eq!(p.upload_window, 8);
assert_eq!(e.phase(), TunePhase::Uninitialized);
assert!(e.tick(1_000.0, 12_500_000, Some(40.0), false).is_none());
assert_eq!(p, static_params().clamped_into(&caps()));
}
}