use std::cell::RefCell;
use std::rc::Rc;
use libfw_core::Capabilities;
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 DEFAULT_TUNE_TTL_MS: u64 = 3_600_000;
pub const CACHE_VERSION: u32 = 1;
pub const MIN_WINDOWS_TO_PERSIST: u32 = 2;
pub const RTT_DRIFT_RE_RAMP: f64 = 0.50;
pub const MAX_CONSECUTIVE_FAILURES: u32 = 3;
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 download_chunk_size: u64,
pub compress_level: i32,
}
impl TuneParams {
pub fn from_caps_mins(caps: &Capabilities) -> TuneParams {
TuneParams {
concurrency: caps.limits.concurrency.min.max(1) as usize,
upload_window: caps.limits.upload_window.min.max(1) as usize,
download_window: caps.limits.download_window.min.max(1) as usize,
chunk_size: caps.limits.chunk_size.min.max(1) as u64,
download_chunk_size: caps.limits.download_chunk_size.min.max(1) as u64,
compress_level: caps.compression.zrip_levels.min,
}
}
pub fn from_config(
concurrency: usize,
upload_window: usize,
download_window: usize,
chunk_size: u64,
download_chunk_size: u64,
compress_level: i32,
caps: &Capabilities,
) -> TuneParams {
TuneParams {
concurrency: caps.limits.concurrency.clamp(concurrency.max(1) as i64).max(1) as usize,
upload_window: caps.limits.upload_window.clamp(upload_window.max(1) as i64).max(1)
as usize,
download_window: caps
.limits
.download_window
.clamp(download_window.max(1) as i64)
.max(1) as usize,
chunk_size: caps.limits.chunk_size.clamp(chunk_size.max(1) as i64).max(1) as u64,
download_chunk_size: caps
.limits
.download_chunk_size
.clamp(download_chunk_size.max(1) as i64)
.max(1) as u64,
compress_level: caps.clamp_level(compress_level),
}
}
pub fn clamped_into(&self, caps: &Capabilities) -> TuneParams {
TuneParams {
concurrency: caps.limits.concurrency.clamp(self.concurrency as i64).max(1) as usize,
upload_window: caps.limits.upload_window.clamp(self.upload_window as i64).max(1)
as usize,
download_window: caps
.limits
.download_window
.clamp(self.download_window as i64)
.max(1) as usize,
chunk_size: caps.limits.chunk_size.clamp(self.chunk_size as i64).max(1) as u64,
download_chunk_size: caps
.limits
.download_chunk_size
.clamp(self.download_chunk_size as i64)
.max(1) 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.download_chunk_size
}
}
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TuneCache {
pub v: u32,
pub ts: u64,
pub caps_hash: String,
pub params: TuneParams,
pub stats: TuneStats,
}
pub trait TuneStore {
fn get(&self, key: &str) -> Option<String>;
fn set(&self, key: &str, value: &str);
fn remove(&self, key: &str);
}
pub struct LocalStore;
impl TuneStore for LocalStore {
fn get(&self, key: &str) -> Option<String> {
web_sys::window()?
.local_storage()
.ok()
.flatten()?
.get_item(key)
.ok()
.flatten()
}
fn set(&self, key: &str, value: &str) {
if let Some(storage) = web_sys::window().and_then(|w| w.local_storage().ok().flatten()) {
let _ = storage.set_item(key, value);
}
}
fn remove(&self, key: &str) {
if let Some(storage) = web_sys::window().and_then(|w| w.local_storage().ok().flatten()) {
let _ = storage.remove_item(key);
}
}
}
pub fn tune_key(base_url: &str) -> String {
format!("libfw.tune.{}", origin_hash(base_url))
}
pub fn origin_hash(base_url: &str) -> String {
use sha2::Digest;
let origin = extract_origin(base_url);
let digest = sha2::Sha256::digest(origin.as_bytes());
format!("sha256-{}", hex_lower(&digest))
}
fn extract_origin(url: &str) -> String {
match url.find("://") {
Some(i) => {
let scheme = &url[..i];
let rest = &url[i + 3..];
let host_port = rest.split('/').next().unwrap_or("");
let host_port = host_port.rsplit('@').next().unwrap_or(host_port);
format!(
"{}://{}",
scheme.to_ascii_lowercase(),
host_port.to_ascii_lowercase()
)
}
None => url.to_ascii_lowercase(),
}
}
fn hex_lower(bytes: &[u8]) -> String {
const HEX: &[u8; 16] = b"0123456789abcdef";
let mut out = String::with_capacity(bytes.len() * 2);
for &b in bytes {
out.push(HEX[(b >> 4) as usize] as char);
out.push(HEX[(b & 0x0f) as usize] as char);
}
out
}
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
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RampDim {
Window,
Concurrency,
ChunkSize,
}
#[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 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 {
1.0
};
if gain >= GAIN_THRESHOLD {
return (RampAction::Raise, 0);
}
let low = input.low_gain_windows + 1;
if low >= 2 {
(RampAction::Settle, low)
} 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,
ttl_ms: u64,
level_cfg: CompressLevel,
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>,
prev_mbps: f64,
low_gain_windows: u32,
dim: RampDim,
degraded_windows: u32,
consecutive_failures: u32,
post_degrade_settle: bool,
direction: TransferKind,
on_tuning: Option<js_sys::Function>,
}
impl TuningEngine {
pub fn new(enabled: bool, ttl_ms: u64, level_cfg: CompressLevel) -> TuningEngine {
TuningEngine {
enabled,
ttl_ms,
level_cfg,
phase: TunePhase::Uninitialized,
caps: None,
caps_hash: String::new(),
params: TuneParams {
concurrency: 1,
upload_window: 1,
download_window: 1,
chunk_size: 1,
download_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,
prev_mbps: 0.0,
low_gain_windows: 0,
dim: RampDim::Window,
degraded_windows: 0,
consecutive_failures: 0,
post_degrade_settle: false,
direction: TransferKind::Download,
on_tuning: None,
}
}
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,
base_url: &str,
caps: &Capabilities,
store: &dyn TuneStore,
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;
if !self.enabled {
self.phase = TunePhase::Uninitialized;
self.params = static_params.clamped_into(caps);
return self.params.clone();
}
let key = tune_key(base_url);
let cached: Option<TuneCache> = store
.get(&key)
.and_then(|s| serde_json::from_str(&s).ok())
.filter(|c: &TuneCache| c.v == CACHE_VERSION);
match cached {
Some(c)
if c.caps_hash == self.caps_hash
&& (now_ms - c.ts as f64) < self.ttl_ms as f64 =>
{
self.params = c.params.clamped_into(caps);
self.stats = c.stats;
self.phase = TunePhase::Settled;
self.dim = RampDim::Window;
}
cached => {
if cached
.filter(|c| c.caps_hash != self.caps_hash)
.is_some()
{
store.remove(&key);
}
self.params = TuneParams::from_caps_mins(caps);
self.stats = TuneStats::default();
self.phase = TunePhase::Ramping;
self.dim = RampDim::Window;
}
}
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 {
let drift_ratio = if self.stats.rtt_ms > 0.0 {
(new_ewma - self.stats.rtt_ms).abs() / self.stats.rtt_ms
} else {
0.0
};
let drift_visible = drift_ratio > RTT_DRIFT_RE_RAMP
&& (self.windows_closed >= 1 || drift_ratio > 1.0);
if let Some(caps) = self.caps.as_ref().filter(|_| {
self.phase == TunePhase::Settled
&& self.stats.rtt_ms > 0.0
&& drift_visible
}) {
self.params = TuneParams::from_caps_mins(caps);
self.stats = TuneStats::default();
self.phase = TunePhase::Ramping;
self.dim = RampDim::Window;
self.low_gain_windows = 0;
}
}
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;
}
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,
low_gain_windows: self.low_gain_windows,
in_flight: self.in_flight(),
bdp: if ramping {
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::ChunkSize,
};
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);
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;
}
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,
base_url: &str,
store: &dyn TuneStore,
now_ms: f64,
ok: bool,
) {
if !self.enabled {
return;
}
if ok {
self.consecutive_failures = 0;
if self.phase == TunePhase::Settled
&& !self.post_degrade_settle
&& self.windows_closed >= MIN_WINDOWS_TO_PERSIST
{
self.persist(base_url, store, now_ms);
}
} else {
self.consecutive_failures += 1;
if self.consecutive_failures >= MAX_CONSECUTIVE_FAILURES {
store.remove(&tune_key(base_url));
self.consecutive_failures = 0;
self.phase = TunePhase::Ramping;
}
}
}
pub fn persist(&self, base_url: &str, store: &dyn TuneStore, now_ms: f64) {
if !self.enabled || self.phase != TunePhase::Settled {
return;
}
let cache = TuneCache {
v: CACHE_VERSION,
ts: now_ms as u64,
caps_hash: self.caps_hash.clone(),
params: self.params.clone(),
stats: self.stats,
};
if let Ok(json) = serde_json::to_string(&cache) {
store.set(&tune_key(base_url), &json);
}
}
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.stats.rtt_ms) {
(Some(cur), base) if base > 0.0 && self.phase == TunePhase::Ramping => {
cur / base - 1.0
}
_ => 0.0,
}
}
fn in_flight(&self) -> u64 {
match self.dim {
RampDim::Window | RampDim::Concurrency | RampDim::ChunkSize => match self
.last_transfer_kind()
{
TransferKind::Upload => self.params.upload_in_flight(),
TransferKind::Download => self.params.download_in_flight(),
},
}
}
fn dim_at_cap(&self, caps: &Capabilities) -> bool {
match self.dim {
RampDim::Window => match self.last_transfer_kind() {
TransferKind::Upload => {
self.params.upload_window as i64 >= caps.limits.upload_window.max
}
TransferKind::Download => {
self.params.download_window as i64 >= caps.limits.download_window.max
}
},
RampDim::Concurrency => {
self.params.concurrency as i64 >= caps.limits.concurrency.max
}
RampDim::ChunkSize => match self.last_transfer_kind() {
TransferKind::Upload => self.params.chunk_size >= caps.limits.chunk_size.max as u64,
TransferKind::Download => {
self.params.download_chunk_size >= caps.limits.download_chunk_size.max as u64
}
},
}
}
fn raise_dim(&mut self, caps: &Capabilities) {
let dim_max: i64 = 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,
RampDim::ChunkSize => match self.last_transfer_kind() {
TransferKind::Upload => caps.limits.chunk_size.max,
TransferKind::Download => caps.limits.download_chunk_size.max,
},
};
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
}
RampDim::ChunkSize => match self.last_transfer_kind() {
TransferKind::Upload => {
self.params.chunk_size = step(self.params.chunk_size as i64).max(1) as u64
}
TransferKind::Download => {
self.params.download_chunk_size =
step(self.params.download_chunk_size as i64).max(1) as u64
}
},
}
}
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
}
RampDim::ChunkSize => match self.last_transfer_kind() {
TransferKind::Upload => {
self.params.chunk_size = step(self.params.chunk_size as i64) as u64
}
TransferKind::Download => {
self.params.download_chunk_size = step(self.params.download_chunk_size as i64)
as u64
}
},
}
}
}
#[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::ChunkSize,
RampDim::ChunkSize => RampDim::ChunkSize,
}
}
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("downloadChunkSize"), &JsValue::from_f64(params.download_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::*;
use std::cell::RefCell;
use std::collections::HashMap;
#[derive(Default)]
struct FakeStore {
map: RefCell<HashMap<String, String>>,
}
impl TuneStore for FakeStore {
fn get(&self, key: &str) -> Option<String> {
self.map.borrow().get(key).cloned()
}
fn set(&self, key: &str, value: &str) {
self.map.borrow_mut().insert(key.to_string(), value.to_string());
}
fn remove(&self, key: &str) {
self.map.borrow_mut().remove(key);
}
}
fn caps() -> Capabilities {
Capabilities::default()
}
fn static_params() -> TuneParams {
TuneParams {
concurrency: 4,
upload_window: 8,
download_window: 4,
chunk_size: 2 * 1024 * 1024,
download_chunk_size: 256 * 1024,
compress_level: 1,
}
}
fn engine(store: &dyn TuneStore) -> TuningEngine {
let mut e = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
e.begin_transfer("https://h.example/", &caps(), store, 0.0, &static_params());
e
}
#[test]
fn ramp_action_error_degrades() {
let input = RampInput {
mbps: 10.0,
prev_mbps: 5.0,
rtt_inflation: 0.0,
errors: 1,
degraded: false,
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,
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,
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,
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::Settle);
assert_eq!(low, 2);
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,
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 origin_hash_is_stable_and_discriminates() {
assert_eq!(origin_hash("https://h.example/"), origin_hash("https://h.example/a/b"));
assert_eq!(origin_hash("https://h.example:8443/x"), origin_hash("https://h.example:8443/y"));
assert_ne!(origin_hash("https://h.example/"), origin_hash("https://h.example:8443/"));
assert_ne!(origin_hash("https://h.example/"), origin_hash("http://h.example/"));
assert_ne!(origin_hash("https://h.example/"), origin_hash("https://other.example/"));
assert!(tune_key("https://h.example/").starts_with("libfw.tune.sha256-"));
}
#[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 store = FakeStore::default();
let mut e = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
let p = e.begin_transfer("https://h.example/", &caps(), &store, 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 ramp_raises_window_then_concurrency_until_cap() {
let store = FakeStore::default();
let mut e = engine(&store);
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 store = FakeStore::default();
let mut e = engine(&store);
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 low_gain_twice_settles_and_persists() {
let store = FakeStore::default();
let mut e = engine(&store);
let ev = e.tick(1_000.0, 12_500_000, Some(400.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::Ramping);
let ev = e.tick(3_000.0, 26_000_000, Some(400.0), false).unwrap(); assert_eq!(ev.phase, TunePhase::Settled);
let ev = e.tick(4_000.0, 27_040_000, Some(400.0), false).unwrap(); assert_eq!(ev.phase, TunePhase::Settled);
assert_eq!(ev.params.upload_window, 2);
e.transfer_end("https://h.example/", &store, 4_000.0, true);
let key = tune_key("https://h.example/");
let cached: TuneCache = serde_json::from_str(&store.get(&key).unwrap()).unwrap();
assert_eq!(cached.params.upload_window, 2);
assert_eq!(cached.caps_hash, caps().caps_hash());
}
#[test]
fn cache_reuse_settles_immediately() {
let store = FakeStore::default();
{
let mut e = engine(&store);
e.tick(1_000.0, 12_500_000, Some(400.0), false);
e.tick(2_000.0, 25_000_000, Some(400.0), false);
e.tick(3_000.0, 26_000_000, Some(400.0), false);
e.tick(4_000.0, 27_040_000, Some(400.0), false);
e.transfer_end("https://h.example/", &store, 4_000.0, true);
}
let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
e2.set_direction(TransferKind::Upload);
let p = e2.begin_transfer("https://h.example/", &caps(), &store, 10_000.0, &static_params());
assert_eq!(e2.phase(), TunePhase::Settled);
assert_eq!(p.upload_window, 2);
}
#[test]
fn ttl_expiry_re_ramps() {
let store = FakeStore::default();
{
let mut e = engine(&store);
e.tick(1_000.0, 12_500_000, Some(400.0), false);
e.tick(2_000.0, 25_000_000, Some(400.0), false);
e.tick(3_000.0, 26_000_000, Some(400.0), false);
e.tick(4_000.0, 27_040_000, Some(400.0), false);
e.transfer_end("https://h.example/", &store, 4_000.0, true);
}
let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
e2.set_direction(TransferKind::Upload);
let p = e2.begin_transfer(
"https://h.example/",
&caps(),
&store,
4_000.0 + DEFAULT_TUNE_TTL_MS as f64 + 1.0,
&static_params(),
);
assert_eq!(e2.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1, "expired cache → ramp from mins");
}
#[test]
fn caps_hash_change_invalidates_cache() {
let store = FakeStore::default();
{
let mut e = engine(&store);
e.tick(1_000.0, 12_500_000, Some(400.0), false);
e.tick(2_000.0, 25_000_000, Some(400.0), false);
e.tick(3_000.0, 26_000_000, Some(400.0), false);
e.tick(4_000.0, 27_040_000, Some(400.0), false);
e.transfer_end("https://h.example/", &store, 4_000.0, true);
}
let mut changed = caps();
changed.limits.concurrency.default = 6;
let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
e2.set_direction(TransferKind::Upload);
let p = e2.begin_transfer("https://h.example/", &changed, &store, 10_000.0, &static_params());
assert_eq!(e2.phase(), TunePhase::Ramping);
assert_eq!(p.upload_window, 1);
assert!(store.get(&tune_key("https://h.example/")).is_none(), "stale cache removed");
}
#[test]
fn short_transfer_does_not_persist() {
let store = FakeStore::default();
let mut e = engine(&store);
e.tick(1_000.0, 12_500_000, Some(40.0), false);
e.transfer_end("https://h.example/", &store, 1_500.0, true);
assert!(store.get(&tune_key("https://h.example/")).is_none(), "short transfers are noise");
}
#[test]
fn repeated_failures_remove_cache() {
let store = FakeStore::default();
let mut e = engine(&store);
e.tick(1_000.0, 12_500_000, Some(40.0), false);
e.tick(2_000.0, 25_000_000, Some(40.0), false);
e.tick(3_000.0, 26_000_000, Some(40.0), false);
e.tick(4_000.0, 27_040_000, Some(40.0), false);
e.transfer_end("https://h.example/", &store, 4_000.0, true);
assert!(store.get(&tune_key("https://h.example/")).is_some());
for _ in 0..3 {
e.transfer_end("https://h.example/", &store, 5_000.0, false);
}
assert!(store.get(&tune_key("https://h.example/")).is_none());
}
#[test]
fn rtt_drift_re_ramps_from_cache() {
let store = FakeStore::default();
{
let mut e = engine(&store);
e.tick(1_000.0, 12_500_000, Some(400.0), false);
e.tick(2_000.0, 25_000_000, Some(400.0), false);
e.tick(3_000.0, 26_000_000, Some(400.0), false);
e.tick(4_000.0, 27_040_000, Some(400.0), false);
e.transfer_end("https://h.example/", &store, 4_000.0, true);
}
let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
e2.set_direction(TransferKind::Upload);
e2.begin_transfer("https://h.example/", &caps(), &store, 10_000.0, &static_params());
assert_eq!(e2.phase(), TunePhase::Settled);
let ev = e2.tick(11_000.0, 12_500_000, Some(1_200.0), false).unwrap();
assert_eq!(ev.phase, TunePhase::Ramping);
assert!(
ev.params.upload_window <= 2,
"re-ramp starts from mins (first measured window may raise once)"
);
}
#[test]
fn upload_auto_level_benchmarks_once_and_caches() {
let store = FakeStore::default();
let mut e = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Auto);
e.begin_transfer("https://h.example/", &caps(), &store, 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, DEFAULT_TUNE_TTL_MS, CompressLevel::Auto);
e2.begin_transfer("https://h.example/", &caps(), &store, 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 store = FakeStore::default();
let mut e = TuningEngine::new(false, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
e.set_direction(TransferKind::Upload);
let p = e.begin_transfer("https://h.example/", &caps(), &store, 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()));
}
}