Skip to main content

libfw_client/
tune.rs

1//! Adaptive tuning engine: TCP-style ramp over *real* transfers.
2//!
3//! # Model
4//!
5//! The engine treats each transfer as a slow-start probe: measurements are
6//! taken from the bytes the transfer actually moves (no synthetic probe
7//! requests), and the shared parameter table is ramped up dimension by
8//! dimension — per-file window first, then cross-file concurrency, then
9//! chunk size — until the link saturates, then settled and persisted to
10//! `localStorage` (via a [`TuneStore`]) for reuse by later transfers.
11//!
12//! # Design notes
13//!
14//! - **Pure decision logic**: [`ramp_action`], [`choose_auto_level`],
15//!   [`origin_hash`] and the clamp/EWMA helpers are plain functions with
16//!   table-driven tests. Everything that touches the clock or storage is
17//!   injected, so native tests never need a browser.
18//! - **Shared state**: one [`TuningEngine`] per client, wrapped in
19//!   `Rc<RefCell<_>>` (WASM is single-threaded). All transfers read the
20//!   current params when they schedule work.
21//! - **One-second windows**: [`TuningEngine::tick`] accumulates bytes / RTT
22//!   samples / errors and evaluates exactly one window per second of wall
23//!   clock, so measurements are comparable across links.
24//! - **Persistence**: settled params are stored under
25//!   `libfw.tune.<originHash>` with a TTL and the server's `capsHash`;
26//!   stale entries, hash mismatches and repeated failures invalidate it.
27
28use std::cell::RefCell;
29use std::rc::Rc;
30
31use libfw_core::Capabilities;
32use serde::{Deserialize, Serialize};
33use wasm_bindgen::JsValue;
34
35/// Measurement window length (ms).
36pub const TUNE_WINDOW_MS: f64 = 1_000.0;
37/// Throughput growth that justifies raising a dimension (≥ +5%).
38pub const GAIN_THRESHOLD: f64 = 0.05;
39/// RTT inflation (vs the ramping baseline) that signals saturation (> 30%).
40pub const RTT_INFLATE_THRESHOLD: f64 = 0.30;
41/// How far a dimension drops on errors (× 0.5).
42pub const DEGRADE_FACTOR: f64 = 0.5;
43/// EWMA smoothing for RTT samples.
44pub const RTT_EWMA_ALPHA: f64 = 0.25;
45/// Default TTL of a persisted tuning cache (1 hour).
46pub const DEFAULT_TUNE_TTL_MS: u64 = 3_600_000;
47/// Cache schema version — bump to invalidate all persisted entries.
48pub const CACHE_VERSION: u32 = 1;
49/// A transfer must have completed at least this many windows before its
50/// settled params are persisted (short transfers are measurement noise).
51pub const MIN_WINDOWS_TO_PERSIST: u32 = 2;
52/// RTT deviation (vs cached stats) that forces a re-ramp (> 50%).
53pub const RTT_DRIFT_RE_RAMP: f64 = 0.50;
54/// Consecutive failed transfers that invalidate the cache.
55pub const MAX_CONSECUTIVE_FAILURES: u32 = 3;
56/// Sample size for the compression-level micro-benchmark.
57///
58/// 64 KiB rather than 256 KiB: the benchmark runs synchronously on the WASM
59/// main thread (no worker threads available), so a large sample would block
60/// the JS event loop for a noticeable period (50 ms+) on slow devices. 64 KiB
61/// is sufficient for zstd's dictionary-learning stage to converge and gives a
62/// representative ratio without the UI stutter. The caller is expected to
63/// pass `sample[..LEVEL_SAMPLE_SIZE.min(sample.len())]`.
64pub const LEVEL_SAMPLE_SIZE: usize = 64 * 1024;
65
66// ---------------------------------------------------------------------------
67// Data model
68// ---------------------------------------------------------------------------
69
70/// Lifecycle of the tuning state machine.
71#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
72#[serde(rename_all = "lowercase")]
73pub enum TunePhase {
74    /// No transfer has run yet in this session.
75    Uninitialized,
76    /// Ramping parameters up from the advertised minimums.
77    Ramping,
78    /// Parameters converged; served from cache on later transfers.
79    Settled,
80    /// An error shrank the parameters; waiting for stable windows.
81    Degraded,
82}
83
84/// Smoothed transfer statistics.
85#[derive(Debug, Clone, Copy, PartialEq, Default, Serialize, Deserialize)]
86#[serde(rename_all = "camelCase")]
87pub struct TuneStats {
88    /// EWMA of chunk-request time-to-first-byte (ms).
89    pub rtt_ms: f64,
90    /// Last-window throughput (Mbps).
91    pub mbps: f64,
92}
93
94/// The shared, live parameter table.
95#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
96#[serde(rename_all = "camelCase")]
97pub struct TuneParams {
98    /// Cross-file fan-out / global in-flight request cap.
99    pub concurrency: usize,
100    /// Per-file upload in-flight chunk window.
101    pub upload_window: usize,
102    /// Per-file download in-flight chunk window.
103    pub download_window: usize,
104    /// Shared transfer chunk size for both uploads and parallel downloads.
105    pub chunk_size: u64,
106    /// zrip level for this transfer's compressed body.
107    pub compress_level: i32,
108}
109
110impl TuneParams {
111    /// The parameter table at the server's advertised minimums (ramp start).
112    pub fn from_caps_mins(caps: &Capabilities) -> TuneParams {
113        TuneParams {
114            concurrency: caps.limits.concurrency.min.max(1) as usize,
115            upload_window: caps.limits.upload_window.min.max(1) as usize,
116            download_window: caps.limits.download_window.min.max(1) as usize,
117            chunk_size: caps.limits.chunk_size.min.max(1) as u64,
118            compress_level: caps.compression.zrip_levels.min,
119        }
120    }
121
122    /// A static table derived from the client configuration (no tuning:
123    /// legacy behavior, but still level-resolved and clamped to the caps).
124    pub fn from_config(
125        concurrency: usize,
126        upload_window: usize,
127        download_window: usize,
128        chunk_size: u64,
129        compress_level: i32,
130        caps: &Capabilities,
131    ) -> TuneParams {
132        TuneParams {
133            concurrency: caps.limits.concurrency.clamp(concurrency.max(1) as i64).max(1) as usize,
134            upload_window: caps.limits.upload_window.clamp(upload_window.max(1) as i64).max(1)
135                as usize,
136            download_window: caps
137                .limits
138                .download_window
139                .clamp(download_window.max(1) as i64)
140                .max(1) as usize,
141            chunk_size: caps.limits.chunk_size.clamp(chunk_size.max(1) as i64).max(1) as u64,
142            compress_level: caps.clamp_level(compress_level),
143        }
144    }
145
146    /// Clamp every dimension into the server's advertised ranges (defensive
147    /// against a server that shrank its caps since we cached params).
148    pub fn clamped_into(&self, caps: &Capabilities) -> TuneParams {
149        TuneParams {
150            concurrency: caps.limits.concurrency.clamp(self.concurrency as i64).max(1) as usize,
151            upload_window: caps.limits.upload_window.clamp(self.upload_window as i64).max(1)
152                as usize,
153            download_window: caps
154                .limits
155                .download_window
156                .clamp(self.download_window as i64)
157                .max(1) as usize,
158            chunk_size: caps.limits.chunk_size.clamp(self.chunk_size as i64).max(1) as u64,
159            compress_level: caps.clamp_level(self.compress_level),
160        }
161    }
162
163    /// In-flight bytes for an upload: concurrency × window × chunk size.
164    pub fn upload_in_flight(&self) -> u64 {
165        self.concurrency as u64 * self.upload_window as u64 * self.chunk_size
166    }
167
168    /// In-flight bytes for a download: concurrency × window × chunk size.
169    pub fn download_in_flight(&self) -> u64 {
170        self.concurrency as u64 * self.download_window as u64 * self.chunk_size
171    }
172}
173
174/// Persisted tuning cache entry (`localStorage["libfw.tune.<originHash>"]`).
175#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
176#[serde(rename_all = "camelCase")]
177pub struct TuneCache {
178    pub v: u32,
179    /// Settle timestamp (epoch ms).
180    pub ts: u64,
181    /// Capabilities hash this entry was tuned against.
182    pub caps_hash: String,
183    pub params: TuneParams,
184    pub stats: TuneStats,
185}
186
187// ---------------------------------------------------------------------------
188// Storage
189// ---------------------------------------------------------------------------
190
191/// Injected persistence backend (localStorage in the browser; a map in tests).
192pub trait TuneStore {
193    fn get(&self, key: &str) -> Option<String>;
194    fn set(&self, key: &str, value: &str);
195    fn remove(&self, key: &str);
196}
197
198/// localStorage-backed store (only reachable in the browser).
199pub struct LocalStore;
200
201impl TuneStore for LocalStore {
202    fn get(&self, key: &str) -> Option<String> {
203        web_sys::window()?
204            .local_storage()
205            .ok()
206            .flatten()?
207            .get_item(key)
208            .ok()
209            .flatten()
210    }
211
212    fn set(&self, key: &str, value: &str) {
213        if let Some(storage) = web_sys::window().and_then(|w| w.local_storage().ok().flatten()) {
214            let _ = storage.set_item(key, value);
215        }
216    }
217
218    fn remove(&self, key: &str) {
219        if let Some(storage) = web_sys::window().and_then(|w| w.local_storage().ok().flatten()) {
220            let _ = storage.remove_item(key);
221        }
222    }
223}
224
225/// The localStorage key for a base URL: `libfw.tune.<originHash>`.
226///
227/// Keyed by **origin** (not base path): the same server advertises the same
228/// capabilities regardless of mount point, so all its paths share one cache.
229pub fn tune_key(base_url: &str) -> String {
230    format!("libfw.tune.{}", origin_hash(base_url))
231}
232
233/// Stable content hash of a base URL's origin (scheme + host + port).
234pub fn origin_hash(base_url: &str) -> String {
235    use sha2::Digest;
236    let origin = extract_origin(base_url);
237    let digest = sha2::Sha256::digest(origin.as_bytes());
238    format!("sha256-{}", hex_lower(&digest))
239}
240
241/// `scheme://host[:port]` (lowercased, userinfo stripped). Pure string work
242/// so native tests exercise it without a browser `URL` object.
243fn extract_origin(url: &str) -> String {
244    match url.find("://") {
245        Some(i) => {
246            let scheme = &url[..i];
247            let rest = &url[i + 3..];
248            let host_port = rest.split('/').next().unwrap_or("");
249            // Strip userinfo (`user:pass@host`) just in case.
250            let host_port = host_port.rsplit('@').next().unwrap_or(host_port);
251            format!(
252                "{}://{}",
253                scheme.to_ascii_lowercase(),
254                host_port.to_ascii_lowercase()
255            )
256        }
257        None => url.to_ascii_lowercase(),
258    }
259}
260
261fn hex_lower(bytes: &[u8]) -> String {
262    const HEX: &[u8; 16] = b"0123456789abcdef";
263    let mut out = String::with_capacity(bytes.len() * 2);
264    for &b in bytes {
265        out.push(HEX[(b >> 4) as usize] as char);
266        out.push(HEX[(b & 0x0f) as usize] as char);
267    }
268    out
269}
270
271// ---------------------------------------------------------------------------
272// Pure helpers
273// ---------------------------------------------------------------------------
274
275/// EWMA of `sample` into `prev` (`None` seeds the first sample).
276pub fn ewma(prev: Option<f64>, sample: f64, alpha: f64) -> f64 {
277    match prev {
278        None => sample,
279        Some(p) => alpha * sample + (1.0 - alpha) * p,
280    }
281}
282
283/// Bandwidth-delay product in bytes from Mbps and RTT ms.
284pub fn bdp_bytes(mbps: f64, rtt_ms: f64) -> u64 {
285    (mbps * 1e6 / 8.0 * rtt_ms / 1_000.0).max(0.0) as u64
286}
287
288/// The dimension currently being raised.
289#[derive(Debug, Clone, Copy, PartialEq, Eq)]
290pub enum RampDim {
291    /// The active transfer's per-file window (upload or download).
292    Window,
293    /// Cross-file concurrency (global in-flight request cap).
294    Concurrency,
295    /// The active transfer's chunk size.
296    ChunkSize,
297}
298
299/// What a window evaluation decided to do.
300#[derive(Debug, Clone, Copy, PartialEq, Eq)]
301pub enum RampAction {
302    /// Raise the current dimension (×2, or +1 near the cap).
303    Raise,
304    /// Current dimension is capped — move to the next one.
305    AdvanceDim,
306    /// The link is saturated (or stable long enough): settle.
307    Settle,
308    /// An error happened: halve the current dimension, phase → Degraded.
309    Degrade,
310    /// Below the gain threshold but not yet conclusive: hold.
311    Hold,
312}
313
314/// Inputs to the ramp decision (one measurement window).
315#[derive(Debug, Clone, Copy)]
316pub struct RampInput {
317    /// Throughput of the window just closed (Mbps).
318    pub mbps: f64,
319    /// Throughput of the previous window (0 on the first window).
320    pub prev_mbps: f64,
321    /// `rtt_ewma / rtt_baseline - 1` (0 when no baseline exists yet).
322    pub rtt_inflation: f64,
323    /// Errors observed in this window.
324    pub errors: u32,
325    /// Engine is in the post-degrade stability hold.
326    pub degraded: bool,
327    /// Consecutive windows below the gain threshold (before this one).
328    pub low_gain_windows: u32,
329    /// Current in-flight bytes (concurrency × window × chunk).
330    pub in_flight: u64,
331    /// Bandwidth-delay product estimate (bytes); 0 = unknown.
332    pub bdp: u64,
333    /// Whether the current dimension is at the server's max.
334    pub at_cap: bool,
335    /// Whether the current dimension is the last one to ramp.
336    pub last_dim: bool,
337}
338
339/// The pure ramp decision for one measurement window.
340///
341/// Returns `(action, new_low_gain_windows)`.
342///
343/// Rules (in priority order):
344/// 1. any error → [`RampAction::Degrade`];
345/// 2. current dimension capped → advance (or settle at the last one);
346/// 3. in-flight ≥ BDP → saturated, settle;
347/// 4. RTT inflated > 30% vs baseline → saturated, settle;
348/// 5. throughput gain ≥ +5% → raise (except during the degraded hold);
349/// 6. gain < 5% for 2 consecutive windows → settle;
350/// 7. otherwise hold.
351pub fn ramp_action(input: &RampInput) -> (RampAction, u32) {
352    if input.errors > 0 {
353        return (RampAction::Degrade, 0);
354    }
355    if input.at_cap {
356        return if input.last_dim {
357            (RampAction::Settle, 0)
358        } else {
359            (RampAction::AdvanceDim, 0)
360        };
361    }
362    if input.bdp > 0 && input.in_flight >= input.bdp {
363        return (RampAction::Settle, 0);
364    }
365    if input.rtt_inflation > RTT_INFLATE_THRESHOLD {
366        return (RampAction::Settle, 0);
367    }
368    let gain = if input.degraded {
369        // Degraded stability hold: the window just saw (near-)zero traffic,
370        // so `prev_mbps` may be 0 — the first-window pseudo-gain below must
371        // not re-raise dimensions we just halved. Hold and settle instead.
372        0.0
373    } else if input.prev_mbps > 0.0 {
374        input.mbps / input.prev_mbps - 1.0
375    } else {
376        // First window: no baseline; treat the mere presence of traffic as
377        // growth so the first raise happens immediately.
378        1.0
379    };
380    if gain >= GAIN_THRESHOLD {
381        return (RampAction::Raise, 0);
382    }
383    let low = input.low_gain_windows + 1;
384    if low >= 2 {
385        (RampAction::Settle, low)
386    } else {
387        (RampAction::Hold, low)
388    }
389}
390
391/// Client-side compression level configuration.
392#[derive(Debug, Clone, Copy, PartialEq, Eq)]
393pub enum CompressLevel {
394    /// Micro-benchmark the advertised range on a real file sample.
395    Auto,
396    /// Cheapest (min) — least CPU.
397    Fast,
398    /// The server's advertised default.
399    Balanced,
400    /// Best ratio (max).
401    Max,
402    /// An explicit level (clamped into the advertised range).
403    Fixed(i32),
404}
405
406/// Resolve a [`CompressLevel`] against the advertised zrip range.
407///
408/// `Auto` without a benchmarkable sample degrades to the server default.
409pub fn resolve_level(level: CompressLevel, caps: &Capabilities) -> i32 {
410    let z = caps.compression.zrip_levels;
411    match level {
412        CompressLevel::Fast => z.min,
413        CompressLevel::Balanced | CompressLevel::Auto => z.default,
414        CompressLevel::Max => z.max,
415        CompressLevel::Fixed(l) => z.clamp_level(l),
416    }
417}
418
419/// Choose the best level from micro-benchmark results.
420///
421/// `candidates`: `(level, compressed_len, compress_ms)` in ascending level
422/// order. Scores each by `saved_transfer_ms - compress_ms`, where
423/// `saved_transfer_ms` converts the saved bytes at the current link speed.
424/// Ties go to the cheaper (lower) level. A sample that compresses to
425/// nothing (already-compressed data) yields the cheapest level so the CPU
426/// is never wasted.
427pub fn choose_auto_level(
428    candidates: &[(i32, usize, f64)],
429    uncompressed: usize,
430    mbps: f64,
431) -> i32 {
432    if candidates.is_empty() {
433        return 0;
434    }
435    // Bytes per ms at the current (or a conservative default) link speed.
436    let bytes_per_ms = if mbps > 0.0 {
437        mbps * 1e6 / 8.0 / 1_000.0
438    } else {
439        // 10 Mbps default: don't over-invest CPU on an unknown link.
440        10e6 / 8.0 / 1_000.0
441    };
442    let max_saved = candidates
443        .iter()
444        .map(|&(_, clen, _)| uncompressed.saturating_sub(clen))
445        .max()
446        .unwrap_or(0);
447    if max_saved == 0 {
448        // Incompressible sample: cheapest level.
449        return candidates.iter().map(|&(l, _, _)| l).min().unwrap_or(0);
450    }
451    let mut best = (candidates[0].0, f64::NEG_INFINITY);
452    for &(lvl, clen, cms) in candidates {
453        let saved = uncompressed.saturating_sub(clen) as f64;
454        let score = saved / bytes_per_ms - cms;
455        // Strict `>`: on ties the earlier (cheaper) candidate wins.
456        if score > best.1 {
457            best = (lvl, score);
458        }
459    }
460    best.0
461}
462
463// ---------------------------------------------------------------------------
464// Engine
465// ---------------------------------------------------------------------------
466
467/// A tuning event emitted when a window evaluation changed the state.
468#[derive(Debug, Clone)]
469pub struct TuneEvent {
470    pub phase: TunePhase,
471    pub params: TuneParams,
472    pub stats: TuneStats,
473}
474
475/// Shared handle to the tuning engine (Rc so clones share one table).
476pub type TuneHandle = Rc<RefCell<TuningEngine>>;
477
478/// The tuning state machine.
479pub struct TuningEngine {
480    enabled: bool,
481    ttl_ms: u64,
482    level_cfg: CompressLevel,
483    phase: TunePhase,
484    caps: Option<Capabilities>,
485    caps_hash: String,
486    params: TuneParams,
487    stats: TuneStats,
488    /// Cached auto-level result for this session (per-file benchmark).
489    auto_level: Option<i32>,
490    /// Whether the auto level was benchmarked this session.
491    auto_level_benchmarked: bool,
492    // Window accumulator.
493    window_start_ms: f64,
494    window_done_base: u64,
495    window_errors: u32,
496    windows_closed: u32,
497    rtt_ewma: Option<f64>,
498    prev_mbps: f64,
499    low_gain_windows: u32,
500    dim: RampDim,
501    degraded_windows: u32,
502    consecutive_failures: u32,
503    /// Whether we settled from a Degraded hold (not from a full ramp).
504    /// When true, `transfer_end` skips cache persistence so a halved
505    /// parameter set is never written as the settled baseline.
506    post_degrade_settle: bool,
507    /// Which direction is ramping (drives window/chunk dimension selection).
508    direction: TransferKind,
509    // JS callback for tuning events (SDK `onTuning`).
510    on_tuning: Option<js_sys::Function>,
511}
512
513impl TuningEngine {
514    /// Create the engine from client configuration.
515    pub fn new(enabled: bool, ttl_ms: u64, level_cfg: CompressLevel) -> TuningEngine {
516        TuningEngine {
517            enabled,
518            ttl_ms,
519            level_cfg,
520            phase: TunePhase::Uninitialized,
521            caps: None,
522            caps_hash: String::new(),
523            // Placeholder; replaced on the first begin_transfer.
524            params: TuneParams {
525                concurrency: 1,
526                upload_window: 1,
527                download_window: 1,
528                chunk_size: 1,
529                compress_level: 0,
530            },
531            stats: TuneStats::default(),
532            auto_level: None,
533            auto_level_benchmarked: false,
534            window_start_ms: 0.0,
535            window_done_base: 0,
536            window_errors: 0,
537            windows_closed: 0,
538            rtt_ewma: None,
539            prev_mbps: 0.0,
540            low_gain_windows: 0,
541            dim: RampDim::Window,
542            degraded_windows: 0,
543            consecutive_failures: 0,
544            post_degrade_settle: false,
545            direction: TransferKind::Download,
546            on_tuning: None,
547        }
548    }
549
550    /// Record which direction the coming transfer tunes (sets which
551    /// window/chunk-size dimensions the ramp moves).
552    pub fn set_direction(&mut self, kind: TransferKind) {
553        self.direction = kind;
554    }
555
556    /// Current direction.
557    pub fn direction(&self) -> TransferKind {
558        self.direction
559    }
560
561    /// Install the JS `onTuning(phase, params, stats)` callback.
562    pub fn set_on_tuning(&mut self, cb: Option<js_sys::Function>) {
563        self.on_tuning = cb;
564    }
565
566    /// Current phase.
567    pub fn phase(&self) -> TunePhase {
568        self.phase
569    }
570
571    /// Current live parameters.
572    pub fn params(&self) -> TuneParams {
573        self.params.clone()
574    }
575
576    /// Current stats.
577    pub fn stats(&self) -> TuneStats {
578        self.stats
579    }
580
581    /// Whether tuning is active (autoTune on).
582    pub fn enabled(&self) -> bool {
583        self.enabled
584    }
585
586    /// Force-disable tuning for the rest of this client's lifetime (used
587    /// when the server has no `/capabilities` route — a legacy peer).
588    pub fn set_enabled(&mut self, enabled: bool) {
589        self.enabled = enabled;
590    }
591
592    /// The capabilities currently loaded for this session.
593    pub fn caps(&self) -> Option<Capabilities> {
594        self.caps.clone()
595    }
596
597    /// Snapshot the engine state as a JS object for the SDK:
598    /// `{ phase, params, stats }`.
599    pub fn state_js(&self) -> JsValue {
600        let o = js_sys::Object::new();
601        let phase = match self.phase {
602            TunePhase::Uninitialized => "uninitialized",
603            TunePhase::Ramping => "ramping",
604            TunePhase::Settled => "settled",
605            TunePhase::Degraded => "degraded",
606        };
607        let _ = js_sys::Reflect::set(&o, &JsValue::from_str("phase"), &JsValue::from_str(phase));
608        let _ = js_sys::Reflect::set(&o, &JsValue::from_str("params"), &params_to_js(&self.params));
609        let stats = js_sys::Object::new();
610        let _ = js_sys::Reflect::set(
611            &stats,
612            &JsValue::from_str("rttMs"),
613            &JsValue::from_f64(self.stats.rtt_ms),
614        );
615        let _ = js_sys::Reflect::set(
616            &stats,
617            &JsValue::from_str("mbps"),
618            &JsValue::from_f64(self.stats.mbps),
619        );
620        let _ = js_sys::Reflect::set(&o, &JsValue::from_str("stats"), &stats.into());
621        let _ = js_sys::Reflect::set(
622            &o,
623            &JsValue::from_str("capsHash"),
624            &JsValue::from_str(&self.caps_hash),
625        );
626        o.into()
627    }
628
629    /// The capabilities hash we are currently tuned against.
630    pub fn caps_hash(&self) -> &str {
631        &self.caps_hash
632    }
633
634    /// Start a transfer: consult the cache, then either reuse (Settled) or
635    /// reset to the advertised minimums (Ramping).
636    ///
637    /// Returns the parameters the transfer should start with. When tuning
638    /// is disabled this returns static config-derived params (legacy), so
639    /// callers can always use the result.
640    pub fn begin_transfer(
641        &mut self,
642        base_url: &str,
643        caps: &Capabilities,
644        store: &dyn TuneStore,
645        now_ms: f64,
646        static_params: &TuneParams,
647    ) -> TuneParams {
648        self.caps = Some(caps.clone());
649        self.caps_hash = caps.caps_hash();
650        self.windows_closed = 0;
651        self.window_errors = 0;
652        self.window_done_base = 0;
653        self.window_start_ms = now_ms;
654        self.auto_level_benchmarked = false;
655        self.auto_level = None;
656        self.post_degrade_settle = false;
657
658        if !self.enabled {
659            self.phase = TunePhase::Uninitialized;
660            self.params = static_params.clamped_into(caps);
661            return self.params.clone();
662        }
663
664        let key = tune_key(base_url);
665        let cached: Option<TuneCache> = store
666            .get(&key)
667            .and_then(|s| serde_json::from_str(&s).ok())
668            .filter(|c: &TuneCache| c.v == CACHE_VERSION);
669
670        match cached {
671            Some(c)
672                if c.caps_hash == self.caps_hash
673                    && (now_ms - c.ts as f64) < self.ttl_ms as f64 =>
674            {
675                // Fresh entry for an unchanged server → reuse, Settled.
676                self.params = c.params.clamped_into(caps);
677                self.stats = c.stats;
678                self.phase = TunePhase::Settled;
679                self.dim = RampDim::Window;
680            }
681            cached => {
682                // Miss / expired / changed caps → ramp from the minimums.
683                if cached
684                    .filter(|c| c.caps_hash != self.caps_hash)
685                    .is_some()
686                {
687                    store.remove(&key);
688                }
689                self.params = TuneParams::from_caps_mins(caps);
690                self.stats = TuneStats::default();
691                self.phase = TunePhase::Ramping;
692                self.dim = RampDim::Window;
693            }
694        }
695        self.params.clone()
696    }
697
698    /// Feed one measurement sample; the engine evaluates exactly one window
699    /// per [`TUNE_WINDOW_MS`] of wall clock and may return a tuning event.
700    ///
701    /// `done_bytes` is the absolute transferred count (the engine computes
702    /// the window delta), `rtt_ms` an optional TTFB sample, `error` whether
703    /// a chunk/round failed in this window.
704    pub fn tick(
705        &mut self,
706        now_ms: f64,
707        done_bytes: u64,
708        rtt_ms: Option<f64>,
709        error: bool,
710    ) -> Option<TuneEvent> {
711        if error {
712            self.window_errors += 1;
713        }
714        if let Some(r) = rtt_ms.filter(|&r| r > 0.0) {
715            let new_ewma = ewma(self.rtt_ewma, r, RTT_EWMA_ALPHA);
716            if self.enabled {
717                // Settled reuse: a freshly measured RTT deviating > 50%
718                // from the *cached* stats means the link changed → force
719                // a re-ramp. Compare against `self.stats.rtt_ms` before
720                // it is overwritten by this window's own reading.
721                //
722                // Guard: on the first window (windows_closed == 0) the EWMA
723                // equals the raw sample and has not been smoothed at all.
724                // Only accept a drift signal on window 0 when the deviation
725                // is substantial enough (> 2×) to be obviously real; for
726                // smaller drifts wait until window 1 so the EWMA has at
727                // least one alpha-blend pass before we compare.
728                let drift_ratio = if self.stats.rtt_ms > 0.0 {
729                    (new_ewma - self.stats.rtt_ms).abs() / self.stats.rtt_ms
730                } else {
731                    0.0
732                };
733                let drift_visible = drift_ratio > RTT_DRIFT_RE_RAMP
734                    && (self.windows_closed >= 1 || drift_ratio > 1.0);
735                if let Some(caps) = self.caps.as_ref().filter(|_| {
736                    self.phase == TunePhase::Settled
737                        && self.stats.rtt_ms > 0.0
738                        && drift_visible
739                }) {
740                    self.params = TuneParams::from_caps_mins(caps);
741                    self.stats = TuneStats::default();
742                    self.phase = TunePhase::Ramping;
743                    self.dim = RampDim::Window;
744                    self.low_gain_windows = 0;
745                }
746            }
747            self.rtt_ewma = Some(new_ewma);
748        }
749        let elapsed = now_ms - self.window_start_ms;
750        if elapsed < TUNE_WINDOW_MS {
751            return None;
752        }
753        if !self.enabled {
754            return None;
755        }
756        let caps = self.caps.clone()?;
757
758        let bytes = done_bytes.saturating_sub(self.window_done_base);
759        let mbps = if elapsed > 0.0 {
760            bytes as f64 / elapsed * 1_000.0 * 8.0 / 1e6
761        } else {
762            0.0
763        };
764        self.stats.mbps = mbps;
765        if let Some(rtt) = self.rtt_ewma {
766            self.stats.rtt_ms = rtt;
767        }
768
769        // BDP / RTT saturation only applies while actively ramping: a
770        // Degraded phase's "stability windows" (or a Settled reuse) must not
771        // be cut short by a tiny-BDP computation on near-zero traffic.
772        let ramping = self.phase == TunePhase::Ramping;
773        let input = RampInput {
774            mbps,
775            prev_mbps: self.prev_mbps,
776            rtt_inflation: if ramping { self.rtt_inflation() } else { 0.0 },
777            errors: self.window_errors,
778            degraded: self.phase == TunePhase::Degraded,
779            low_gain_windows: self.low_gain_windows,
780            in_flight: self.in_flight(),
781            bdp: if ramping {
782                bdp_bytes(mbps, self.rtt_ewma.unwrap_or(self.stats.rtt_ms))
783            } else {
784                0
785            },
786            at_cap: self.dim_at_cap(&caps),
787            last_dim: self.dim == RampDim::ChunkSize,
788        };
789        let (action, low) = ramp_action(&input);
790        self.low_gain_windows = low;
791        self.windows_closed += 1;
792        self.prev_mbps = mbps;
793
794        match action {
795            RampAction::Raise => {
796                self.raise_dim(&caps);
797                if self.phase == TunePhase::Degraded {
798                    self.phase = TunePhase::Ramping;
799                }
800            }
801            RampAction::AdvanceDim => {
802                self.dim = next_dim(self.dim);
803            }
804            RampAction::Settle => {
805                self.phase = TunePhase::Settled;
806                self.low_gain_windows = 0;
807            }
808            RampAction::Degrade => {
809                self.halve_dim();
810                self.phase = TunePhase::Degraded;
811                self.degraded_windows = 0;
812                self.low_gain_windows = 0;
813            }
814            RampAction::Hold => {
815                if self.phase == TunePhase::Degraded {
816                    // Conservative settle after 2 stable windows.
817                    self.degraded_windows += 1;
818                    if self.degraded_windows >= 2 {
819                        self.phase = TunePhase::Settled;
820                        // Mark that we settled from a degraded hold so that
821                        // `transfer_end` does NOT persist the reduced params
822                        // as the tuned baseline — the next ramp will
823                        // re-explore from the minimums and cache only after
824                        // a full convergence.
825                        self.post_degrade_settle = true;
826                    }
827                }
828            }
829        }
830
831        // Reset the window accumulator.
832        self.window_start_ms = now_ms;
833        self.window_done_base = done_bytes;
834        self.window_errors = 0;
835
836        let event = TuneEvent {
837            phase: self.phase,
838            params: self.params.clone(),
839            stats: self.stats,
840        };
841        self.emit(&event);
842        Some(event)
843    }
844
845    /// Mark the transfer outcome. On success with a Settled phase and
846    /// enough windows, persist the cache; on failure, count toward cache
847    /// invalidation.
848    pub fn transfer_end(
849        &mut self,
850        base_url: &str,
851        store: &dyn TuneStore,
852        now_ms: f64,
853        ok: bool,
854    ) {
855        if !self.enabled {
856            return;
857        }
858        if ok {
859            self.consecutive_failures = 0;
860            // Only persist a genuine full-ramp settle, not a conservative
861            // post-degrade settle (which ends with halved params — persisting
862            // those would make every subsequent transfer start below the
863            // optimal point and skip the ramp entirely).
864            if self.phase == TunePhase::Settled
865                && !self.post_degrade_settle
866                && self.windows_closed >= MIN_WINDOWS_TO_PERSIST
867            {
868                self.persist(base_url, store, now_ms);
869            }
870        } else {
871            self.consecutive_failures += 1;
872            if self.consecutive_failures >= MAX_CONSECUTIVE_FAILURES {
873                store.remove(&tune_key(base_url));
874                self.consecutive_failures = 0;
875                self.phase = TunePhase::Ramping;
876            }
877        }
878    }
879
880    /// Persist the current settled state.
881    pub fn persist(&self, base_url: &str, store: &dyn TuneStore, now_ms: f64) {
882        if !self.enabled || self.phase != TunePhase::Settled {
883            return;
884        }
885        let cache = TuneCache {
886            v: CACHE_VERSION,
887            ts: now_ms as u64,
888            caps_hash: self.caps_hash.clone(),
889            params: self.params.clone(),
890            stats: self.stats,
891        };
892        if let Ok(json) = serde_json::to_string(&cache) {
893            store.set(&tune_key(base_url), &json);
894        }
895    }
896
897    /// The upload compression level for this session.
898    ///
899    /// `Auto` benchmarks the advertised candidates against `sample` once per
900    /// session (reusing the fastest cache afterwards); without a sample it
901    /// resolves to the server default.
902    pub fn upload_compress_level(
903        &mut self,
904        caps: &Capabilities,
905        sample: Option<&[u8]>,
906        mbps: f64,
907    ) -> i32 {
908        match self.level_cfg {
909            CompressLevel::Auto => {
910                if !self.auto_level_benchmarked {
911                    self.auto_level_benchmarked = true;
912                    self.auto_level = Some(match sample {
913                        Some(s) if !s.is_empty() => {
914                            // Limit the sample to LEVEL_SAMPLE_SIZE so the
915                            // synchronous benchmark does not stall the WASM
916                            // main thread for too long on large files.
917                            let s = &s[..LEVEL_SAMPLE_SIZE.min(s.len())];
918                            let candidates = auto_candidates(caps);
919                            let results = benchmark_levels(s, &candidates);
920                            choose_auto_level(&results, s.len(), mbps.max(1.0))
921                        }
922                        _ => caps.compression.zrip_levels.default,
923                    });
924                }
925                self.auto_level.unwrap_or(caps.compression.zrip_levels.default)
926            }
927            other => resolve_level(other, caps),
928        }
929    }
930
931    /// Whether the auto level has been benchmarked this session.
932    pub fn auto_level_ready(&self) -> bool {
933        self.auto_level_benchmarked
934    }
935
936    fn rtt_inflation(&self) -> f64 {
937        match (self.rtt_ewma, self.stats.rtt_ms) {
938            (Some(cur), base) if base > 0.0 && self.phase == TunePhase::Ramping => {
939                cur / base - 1.0
940            }
941            _ => 0.0,
942        }
943    }
944
945    fn in_flight(&self) -> u64 {
946        match self.dim {
947            RampDim::Window | RampDim::Concurrency | RampDim::ChunkSize => match self
948                .last_transfer_kind()
949            {
950                TransferKind::Upload => self.params.upload_in_flight(),
951                TransferKind::Download => self.params.download_in_flight(),
952            },
953        }
954    }
955
956    fn dim_at_cap(&self, caps: &Capabilities) -> bool {
957        match self.dim {
958            RampDim::Window => match self.last_transfer_kind() {
959                TransferKind::Upload => {
960                    self.params.upload_window as i64 >= caps.limits.upload_window.max
961                }
962                TransferKind::Download => {
963                    self.params.download_window as i64 >= caps.limits.download_window.max
964                }
965            },
966            RampDim::Concurrency => {
967                self.params.concurrency as i64 >= caps.limits.concurrency.max
968            }
969            RampDim::ChunkSize => self.params.chunk_size >= caps.limits.chunk_size.max as u64,
970        }
971    }
972
973    fn raise_dim(&mut self, caps: &Capabilities) {
974        let dim_max: i64 = match self.dim {
975            RampDim::Window => match self.last_transfer_kind() {
976                TransferKind::Upload => caps.limits.upload_window.max,
977                TransferKind::Download => caps.limits.download_window.max,
978            },
979            RampDim::Concurrency => caps.limits.concurrency.max,
980            RampDim::ChunkSize => caps.limits.chunk_size.max,
981        };
982        // Multiplicative increase (×2) until near the cap, then additive
983        // (+1) — TCP slow-start style.
984        let step = |v: i64| -> i64 {
985            if v <= 0 {
986                1
987            } else if v * 2 <= dim_max {
988                v * 2
989            } else {
990                (v + 1).min(dim_max)
991            }
992        };
993        match self.dim {
994            RampDim::Window => match self.last_transfer_kind() {
995                TransferKind::Upload => {
996                    self.params.upload_window = step(self.params.upload_window as i64).max(1) as usize
997                }
998                TransferKind::Download => {
999                    self.params.download_window =
1000                        step(self.params.download_window as i64).max(1) as usize
1001                }
1002            },
1003            RampDim::Concurrency => {
1004                self.params.concurrency = step(self.params.concurrency as i64).max(1) as usize
1005            }
1006            RampDim::ChunkSize => {
1007                self.params.chunk_size = step(self.params.chunk_size as i64).max(1) as u64
1008            }
1009        }
1010    }
1011
1012    fn halve_dim(&mut self) {
1013        let step = |v: i64| -> i64 { ((v as f64) * DEGRADE_FACTOR).floor().max(1.0) as i64 };
1014        match self.dim {
1015            RampDim::Window => match self.last_transfer_kind() {
1016                TransferKind::Upload => {
1017                    self.params.upload_window = step(self.params.upload_window as i64) as usize
1018                }
1019                TransferKind::Download => {
1020                    self.params.download_window = step(self.params.download_window as i64) as usize
1021                }
1022            },
1023            RampDim::Concurrency => {
1024                self.params.concurrency = step(self.params.concurrency as i64) as usize
1025            }
1026            RampDim::ChunkSize => {
1027                self.params.chunk_size = step(self.params.chunk_size as i64) as u64
1028            }
1029        }
1030    }
1031}
1032
1033/// Which transfer direction is ramping (drives which window/chunk dims move).
1034#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1035pub enum TransferKind {
1036    Upload,
1037    Download,
1038}
1039
1040impl TuningEngine {
1041    fn last_transfer_kind(&self) -> TransferKind {
1042        // The ramp dimension set is reset per transfer; the direction is
1043        // recorded by begin_transfer callers via `set_direction`.
1044        self.direction
1045    }
1046}
1047
1048/// The ramp order: per-file window → concurrency → chunk size.
1049fn next_dim(dim: RampDim) -> RampDim {
1050    match dim {
1051        RampDim::Window => RampDim::Concurrency,
1052        RampDim::Concurrency => RampDim::ChunkSize,
1053        // ChunkSize is the terminal dimension: `ramp_action` returns
1054        // `Settle` (not `AdvanceDim`) when it is at cap, so this arm is
1055        // unreachable in practice. It is kept for exhaustiveness; if a new
1056        // dimension is added, the compiler will force updating this function.
1057        RampDim::ChunkSize => RampDim::ChunkSize,
1058    }
1059}
1060
1061/// Benchmark candidates: min, default, max of the advertised range.
1062fn auto_candidates(caps: &Capabilities) -> Vec<i32> {
1063    let z = caps.compression.zrip_levels;
1064    vec![z.min, z.default, z.max]
1065}
1066
1067/// Compress `sample` at each level, returning (level, len, ms).
1068///
1069/// One-shot `zrip::compress` (single frame) is fine for a 256 KiB sample.
1070fn benchmark_levels(sample: &[u8], levels: &[i32]) -> Vec<(i32, usize, f64)> {
1071    levels
1072        .iter()
1073        .map(|&level| {
1074            let t0 = now_ms();
1075            let out = zrip::compress(sample, level).unwrap_or_else(|_| sample.to_vec());
1076            (level, out.len(), now_ms() - t0)
1077        })
1078        .collect()
1079}
1080
1081/// Wall clock (epoch ms). WASM: `Date.now()`; native: `std::time`.
1082pub fn now_ms() -> f64 {
1083    #[cfg(target_arch = "wasm32")]
1084    {
1085        js_sys::Date::now()
1086    }
1087    #[cfg(not(target_arch = "wasm32"))]
1088    {
1089        std::time::SystemTime::now()
1090            .duration_since(std::time::UNIX_EPOCH)
1091            .map(|d| d.as_secs_f64() * 1_000.0)
1092            .unwrap_or(0.0)
1093    }
1094}
1095
1096impl TuningEngine {
1097    fn emit(&self, event: &TuneEvent) {
1098        if let Some(cb) = &self.on_tuning {
1099            let params = params_to_js(&event.params);
1100            let stats = js_sys::Object::new();
1101            let _ = js_sys::Reflect::set(
1102                &stats,
1103                &JsValue::from_str("rttMs"),
1104                &JsValue::from_f64(event.stats.rtt_ms),
1105            );
1106            let _ = js_sys::Reflect::set(
1107                &stats,
1108                &JsValue::from_str("mbps"),
1109                &JsValue::from_f64(event.stats.mbps),
1110            );
1111            let phase = match event.phase {
1112                TunePhase::Uninitialized => "uninitialized",
1113                TunePhase::Ramping => "ramping",
1114                TunePhase::Settled => "settled",
1115                TunePhase::Degraded => "degraded",
1116            };
1117            let _ = cb.call3(
1118                &JsValue::NULL,
1119                &JsValue::from_str(phase),
1120                &params,
1121                &stats.into(),
1122            );
1123        }
1124    }
1125}
1126
1127fn params_to_js(params: &TuneParams) -> JsValue {
1128    let o = js_sys::Object::new();
1129    let _ = js_sys::Reflect::set(&o, &JsValue::from_str("concurrency"), &JsValue::from_f64(params.concurrency as f64));
1130    let _ = js_sys::Reflect::set(&o, &JsValue::from_str("uploadWindow"), &JsValue::from_f64(params.upload_window as f64));
1131    let _ = js_sys::Reflect::set(&o, &JsValue::from_str("downloadWindow"), &JsValue::from_f64(params.download_window as f64));
1132    let _ = js_sys::Reflect::set(&o, &JsValue::from_str("chunkSize"), &JsValue::from_f64(params.chunk_size as f64));
1133    let _ = js_sys::Reflect::set(&o, &JsValue::from_str("compressLevel"), &JsValue::from_f64(params.compress_level as f64));
1134    o.into()
1135}
1136
1137#[cfg(test)]
1138mod tests {
1139    use super::*;
1140    use std::cell::RefCell;
1141    use std::collections::HashMap;
1142
1143    #[derive(Default)]
1144    struct FakeStore {
1145        map: RefCell<HashMap<String, String>>,
1146    }
1147
1148    impl TuneStore for FakeStore {
1149        fn get(&self, key: &str) -> Option<String> {
1150            self.map.borrow().get(key).cloned()
1151        }
1152        fn set(&self, key: &str, value: &str) {
1153            self.map.borrow_mut().insert(key.to_string(), value.to_string());
1154        }
1155        fn remove(&self, key: &str) {
1156            self.map.borrow_mut().remove(key);
1157        }
1158    }
1159
1160    fn caps() -> Capabilities {
1161        Capabilities::default()
1162    }
1163
1164    fn static_params() -> TuneParams {
1165        TuneParams {
1166            concurrency: 4,
1167            upload_window: 8,
1168            download_window: 4,
1169            chunk_size: 2 * 1024 * 1024,
1170            compress_level: 1,
1171        }
1172    }
1173
1174    fn engine(store: &dyn TuneStore) -> TuningEngine {
1175        let mut e = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
1176        e.set_direction(TransferKind::Upload);
1177        e.begin_transfer("https://h.example/", &caps(), store, 0.0, &static_params());
1178        e
1179    }
1180
1181    // --- pure decision logic ------------------------------------------------
1182
1183    #[test]
1184    fn ramp_action_error_degrades() {
1185        let input = RampInput {
1186            mbps: 10.0,
1187            prev_mbps: 5.0,
1188            rtt_inflation: 0.0,
1189            errors: 1,
1190            degraded: false,
1191            low_gain_windows: 0,
1192            in_flight: 100,
1193            bdp: 1_000,
1194            at_cap: false,
1195            last_dim: false,
1196        };
1197        assert_eq!(ramp_action(&input).0, RampAction::Degrade);
1198    }
1199
1200    #[test]
1201    fn ramp_action_cap_advances_or_settles() {
1202        let base = RampInput {
1203            mbps: 10.0,
1204            prev_mbps: 5.0,
1205            rtt_inflation: 0.0,
1206            errors: 0,
1207            degraded: false,
1208            low_gain_windows: 0,
1209            in_flight: 100,
1210            bdp: 1_000,
1211            at_cap: true,
1212            last_dim: false,
1213        };
1214        assert_eq!(ramp_action(&base).0, RampAction::AdvanceDim);
1215        let last = RampInput { last_dim: true, ..base };
1216        assert_eq!(ramp_action(&last).0, RampAction::Settle);
1217    }
1218
1219    #[test]
1220    fn ramp_action_bdp_and_rtt_saturate() {
1221        let base = RampInput {
1222            mbps: 10.0,
1223            prev_mbps: 5.0,
1224            rtt_inflation: 0.0,
1225            errors: 0,
1226            degraded: false,
1227            low_gain_windows: 0,
1228            in_flight: 2_000,
1229            bdp: 1_000,
1230            at_cap: false,
1231            last_dim: false,
1232        };
1233        assert_eq!(ramp_action(&base).0, RampAction::Settle, "in-flight ≥ BDP settles");
1234
1235        let rtt = RampInput {
1236            rtt_inflation: 0.5,
1237            in_flight: 10,
1238            bdp: 1_000,
1239            ..base
1240        };
1241        assert_eq!(ramp_action(&rtt).0, RampAction::Settle, "RTT inflation settles");
1242    }
1243
1244    #[test]
1245    fn ramp_action_gain_raises_and_low_gain_settles() {
1246        // +100% growth → raise (and the low-gain counter resets).
1247        let growing = RampInput {
1248            mbps: 20.0,
1249            prev_mbps: 10.0,
1250            ..RampInput {
1251                mbps: 0.0,
1252                prev_mbps: 0.0,
1253                rtt_inflation: 0.0,
1254                errors: 0,
1255                degraded: false,
1256                low_gain_windows: 7,
1257                in_flight: 10,
1258                bdp: 1_000,
1259                at_cap: false,
1260                last_dim: false,
1261            }
1262        };
1263        let (action, low) = ramp_action(&growing);
1264        assert_eq!(action, RampAction::Raise);
1265        assert_eq!(low, 0, "raise resets the low-gain counter");
1266
1267        // +4% growth (below the 5% threshold) → hold, counter 1.
1268        let flat = RampInput {
1269            mbps: 10.4,
1270            prev_mbps: 10.0,
1271            low_gain_windows: 0,
1272            ..growing
1273        };
1274        let (action, low) = ramp_action(&flat);
1275        assert_eq!(action, RampAction::Hold);
1276        assert_eq!(low, 1, "first below-threshold window holds");
1277
1278        // Second below-threshold window → settle.
1279        let flat2 = RampInput { low_gain_windows: 1, ..flat };
1280        let (action, low) = ramp_action(&flat2);
1281        assert_eq!(action, RampAction::Settle);
1282        assert_eq!(low, 2);
1283
1284        // First window (no baseline) counts as growth.
1285        let first = RampInput {
1286            mbps: 1.0,
1287            prev_mbps: 0.0,
1288            low_gain_windows: 0,
1289            ..growing
1290        };
1291        assert_eq!(ramp_action(&first).0, RampAction::Raise);
1292    }
1293
1294    #[test]
1295    fn ramp_action_degraded_hold_never_raises() {
1296        // Near-zero traffic right after a degrade (prev_mbps == 0) must not
1297        // trigger the first-window pseudo-gain: the engine holds, then
1298        // settles on the second stable window instead of re-raising.
1299        let degraded = RampInput {
1300            mbps: 0.008,
1301            prev_mbps: 0.0,
1302            degraded: true,
1303            ..RampInput {
1304                mbps: 0.0,
1305                prev_mbps: 0.0,
1306                rtt_inflation: 0.0,
1307                errors: 0,
1308                degraded: false,
1309                low_gain_windows: 0,
1310                in_flight: 10,
1311                bdp: 1_000,
1312                at_cap: false,
1313                last_dim: false,
1314            }
1315        };
1316        let (action, low) = ramp_action(&degraded);
1317        assert_eq!(action, RampAction::Hold);
1318        assert_eq!(low, 1, "first stability window holds");
1319        let (action, low) = ramp_action(&RampInput { low_gain_windows: 1, ..degraded });
1320        assert_eq!(action, RampAction::Settle);
1321        assert_eq!(low, 2, "second stability window settles");
1322    }
1323
1324    #[test]
1325    fn ewma_seeds_and_smooths() {
1326        assert_eq!(ewma(None, 10.0, 0.25), 10.0);
1327        let v = ewma(Some(10.0), 20.0, 0.25);
1328        assert!((v - 12.5).abs() < 1e-9);
1329    }
1330
1331    #[test]
1332    fn bdp_conversion() {
1333        // 10 Mbps, 100 ms RTT → 10e6/8 * 0.1 = 125_000 bytes.
1334        assert_eq!(bdp_bytes(10.0, 100.0), 125_000);
1335        assert_eq!(bdp_bytes(0.0, 0.0), 0);
1336    }
1337
1338    #[test]
1339    fn origin_hash_is_stable_and_discriminates() {
1340        assert_eq!(origin_hash("https://h.example/"), origin_hash("https://h.example/a/b"));
1341        assert_eq!(origin_hash("https://h.example:8443/x"), origin_hash("https://h.example:8443/y"));
1342        assert_ne!(origin_hash("https://h.example/"), origin_hash("https://h.example:8443/"));
1343        assert_ne!(origin_hash("https://h.example/"), origin_hash("http://h.example/"));
1344        assert_ne!(origin_hash("https://h.example/"), origin_hash("https://other.example/"));
1345        assert!(tune_key("https://h.example/").starts_with("libfw.tune.sha256-"));
1346    }
1347
1348    #[test]
1349    fn resolve_level_mapping() {
1350        let c = caps();
1351        assert_eq!(resolve_level(CompressLevel::Fast, &c), -8);
1352        assert_eq!(resolve_level(CompressLevel::Balanced, &c), 1);
1353        assert_eq!(resolve_level(CompressLevel::Max, &c), 4);
1354        assert_eq!(resolve_level(CompressLevel::Auto, &c), 1);
1355        assert_eq!(resolve_level(CompressLevel::Fixed(2), &c), 2);
1356        assert_eq!(resolve_level(CompressLevel::Fixed(99), &c), 4);
1357        assert_eq!(resolve_level(CompressLevel::Fixed(-99), &c), -8);
1358    }
1359
1360    #[test]
1361    fn choose_auto_level_picks_by_net_gain() {
1362        // Incompressible sample → cheapest level (min), never waste CPU.
1363        assert_eq!(
1364            choose_auto_level(&[(-8, 1000, 0.1), (1, 1000, 5.0), (4, 1000, 40.0)], 1000, 10.0),
1365            -8
1366        );
1367
1368        // CPU cost dominates (fast link) → cheaper level wins.
1369        let cpu_bound = choose_auto_level(&[(-8, 800_000, 0.1), (4, 400_000, 30.0)], 1_000_000, 1000.0);
1370        assert_eq!(cpu_bound, -8);
1371
1372        // Savings dominate (slow link, cheap CPU) → stronger level wins.
1373        let savings = choose_auto_level(&[(-8, 800_000, 0.1), (4, 400_000, 30.0)], 1_000_000, 10.0);
1374        assert_eq!(savings, 4);
1375
1376        // Ties → cheaper (first) candidate.
1377        assert_eq!(choose_auto_level(&[(0, 500, 1.0), (4, 500, 1.0)], 1000, 100.0), 0);
1378    }
1379
1380    // --- engine behaviour ---------------------------------------------------
1381
1382    #[test]
1383    fn begin_transfer_miss_ramps_from_mins() {
1384        let store = FakeStore::default();
1385        let mut e = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
1386        e.set_direction(TransferKind::Upload);
1387        let p = e.begin_transfer("https://h.example/", &caps(), &store, 0.0, &static_params());
1388        assert_eq!(e.phase(), TunePhase::Ramping);
1389        assert_eq!(p.concurrency, 1);
1390        assert_eq!(p.upload_window, 1);
1391        assert_eq!(p.chunk_size, 262_144);
1392    }
1393
1394    #[test]
1395    fn ramp_raises_window_then_concurrency_until_cap() {
1396        let store = FakeStore::default();
1397        let mut e = engine(&store);
1398        // High-latency profile (400 ms RTT): BDP ≈ 5 MB at 100 Mbps, so the
1399        // ramp can climb the whole window dimension before saturating.
1400        // Deltas double every window (12.5 → 25 → 50 MB/s) → +100% gain.
1401        let mut t = 1_000.0;
1402        let mut done = 12_500_000u64;
1403        let mut delta = 12_500_000u64;
1404        for _ in 0..3 {
1405            let ev = e.tick(t, done, Some(400.0), false).expect("window closed");
1406            assert_eq!(ev.phase, TunePhase::Ramping);
1407            t += 1_000.0;
1408            delta *= 2;
1409            done += delta; // cumulative: 12.5M → 37.5M → 87.5M → ...
1410        }
1411        // Window: 1 → 2 → 4 → 8 (max). Next raise hits the cap → advance.
1412        assert_eq!(e.params().upload_window, 8);
1413        let ev = e.tick(t, done, Some(400.0), false).unwrap();
1414        assert_eq!(ev.params.upload_window, 8); // capped
1415        t += 1_000.0;
1416        delta *= 2;
1417        done += delta;
1418        // Concurrency now ramps: 1 → 2...
1419        let ev = e.tick(t, done, Some(400.0), false).unwrap();
1420        assert_eq!(ev.params.concurrency, 2);
1421        // ...and continues while the window dimension stays put.
1422        let ev = e.tick(t + 1_000.0, done + delta * 2, Some(400.0), false).unwrap();
1423        assert_eq!(ev.params.concurrency, 4);
1424        assert_eq!(ev.params.upload_window, 8);
1425    }
1426
1427    #[test]
1428    fn error_window_degrades_and_halves() {
1429        let store = FakeStore::default();
1430        let mut e = engine(&store);
1431        // Ramp the window up first (400 ms RTT keeps BDP above in-flight).
1432        let mut t = 1_000.0;
1433        let mut done = 12_500_000u64;
1434        let mut delta = 12_500_000u64;
1435        for _ in 0..2 {
1436            e.tick(t, done, Some(400.0), false);
1437            t += 1_000.0;
1438            delta *= 2;
1439            done += delta; // 12.5M → 37.5M: deltas double → +100% gain
1440        }
1441        assert_eq!(e.params().upload_window, 4);
1442        // Error window → degrade: 4 × 0.5 = 2.
1443        let ev = e.tick(t, done, Some(400.0), true).unwrap();
1444        assert_eq!(ev.phase, TunePhase::Degraded);
1445        assert_eq!(ev.params.upload_window, 2);
1446        // Two stable windows → conservative settle (BDP is NOT evaluated in
1447        // the degraded hold — near-zero traffic must not look "saturated").
1448        let ev = e.tick(t + 1_000.0, done + 1_000, None, false).unwrap();
1449        assert_eq!(ev.phase, TunePhase::Degraded);
1450        let ev = e.tick(t + 2_000.0, done + 2_000, None, false).unwrap();
1451        assert_eq!(ev.phase, TunePhase::Settled);
1452        assert_eq!(ev.params.upload_window, 2);
1453    }
1454
1455    #[test]
1456    fn low_gain_twice_settles_and_persists() {
1457        let store = FakeStore::default();
1458        let mut e = engine(&store);
1459        // Raise once, then flatten the throughput → 2 flat windows settle
1460        // (window 1 = hold, window 2 = settle).
1461        let ev = e.tick(1_000.0, 12_500_000, Some(400.0), false).unwrap();
1462        assert_eq!(ev.phase, TunePhase::Ramping);
1463        let ev = e.tick(2_000.0, 25_000_000, Some(400.0), false).unwrap(); // +0% → hold
1464        assert_eq!(ev.phase, TunePhase::Ramping);
1465        let ev = e.tick(3_000.0, 26_000_000, Some(400.0), false).unwrap(); // 2nd flat window
1466        assert_eq!(ev.phase, TunePhase::Settled);
1467        let ev = e.tick(4_000.0, 27_040_000, Some(400.0), false).unwrap(); // stays settled
1468        assert_eq!(ev.phase, TunePhase::Settled);
1469        assert_eq!(ev.params.upload_window, 2);
1470
1471        // ≥2 windows closed & settled → transfer_end persists.
1472        e.transfer_end("https://h.example/", &store, 4_000.0, true);
1473        let key = tune_key("https://h.example/");
1474        let cached: TuneCache = serde_json::from_str(&store.get(&key).unwrap()).unwrap();
1475        assert_eq!(cached.params.upload_window, 2);
1476        assert_eq!(cached.caps_hash, caps().caps_hash());
1477    }
1478
1479    #[test]
1480    fn cache_reuse_settles_immediately() {
1481        let store = FakeStore::default();
1482        // First session: ramp to settle and persist.
1483        {
1484            let mut e = engine(&store);
1485            e.tick(1_000.0, 12_500_000, Some(400.0), false);
1486            e.tick(2_000.0, 25_000_000, Some(400.0), false);
1487            e.tick(3_000.0, 26_000_000, Some(400.0), false);
1488            e.tick(4_000.0, 27_040_000, Some(400.0), false);
1489            e.transfer_end("https://h.example/", &store, 4_000.0, true);
1490        }
1491        // Second session (fresh engine, same store): reuse → Settled.
1492        let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
1493        e2.set_direction(TransferKind::Upload);
1494        let p = e2.begin_transfer("https://h.example/", &caps(), &store, 10_000.0, &static_params());
1495        assert_eq!(e2.phase(), TunePhase::Settled);
1496        assert_eq!(p.upload_window, 2);
1497    }
1498
1499    #[test]
1500    fn ttl_expiry_re_ramps() {
1501        let store = FakeStore::default();
1502        {
1503            let mut e = engine(&store);
1504            e.tick(1_000.0, 12_500_000, Some(400.0), false);
1505            e.tick(2_000.0, 25_000_000, Some(400.0), false);
1506            e.tick(3_000.0, 26_000_000, Some(400.0), false);
1507            e.tick(4_000.0, 27_040_000, Some(400.0), false);
1508            e.transfer_end("https://h.example/", &store, 4_000.0, true);
1509        }
1510        // TTL = 1h; begin at now = 4_000 + 3_600_001 → expired.
1511        let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
1512        e2.set_direction(TransferKind::Upload);
1513        let p = e2.begin_transfer(
1514            "https://h.example/",
1515            &caps(),
1516            &store,
1517            4_000.0 + DEFAULT_TUNE_TTL_MS as f64 + 1.0,
1518            &static_params(),
1519        );
1520        assert_eq!(e2.phase(), TunePhase::Ramping);
1521        assert_eq!(p.upload_window, 1, "expired cache → ramp from mins");
1522    }
1523
1524    #[test]
1525    fn caps_hash_change_invalidates_cache() {
1526        let store = FakeStore::default();
1527        {
1528            let mut e = engine(&store);
1529            e.tick(1_000.0, 12_500_000, Some(400.0), false);
1530            e.tick(2_000.0, 25_000_000, Some(400.0), false);
1531            e.tick(3_000.0, 26_000_000, Some(400.0), false);
1532            e.tick(4_000.0, 27_040_000, Some(400.0), false);
1533            e.transfer_end("https://h.example/", &store, 4_000.0, true);
1534        }
1535        // A server that changed its caps (different hash) → cache removed.
1536        let mut changed = caps();
1537        changed.limits.concurrency.default = 6;
1538        let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
1539        e2.set_direction(TransferKind::Upload);
1540        let p = e2.begin_transfer("https://h.example/", &changed, &store, 10_000.0, &static_params());
1541        assert_eq!(e2.phase(), TunePhase::Ramping);
1542        assert_eq!(p.upload_window, 1);
1543        assert!(store.get(&tune_key("https://h.example/")).is_none(), "stale cache removed");
1544    }
1545
1546    #[test]
1547    fn short_transfer_does_not_persist() {
1548        let store = FakeStore::default();
1549        let mut e = engine(&store);
1550        // Only ONE window closed before the transfer ends.
1551        e.tick(1_000.0, 12_500_000, Some(40.0), false);
1552        e.transfer_end("https://h.example/", &store, 1_500.0, true);
1553        assert!(store.get(&tune_key("https://h.example/")).is_none(), "short transfers are noise");
1554    }
1555
1556    #[test]
1557    fn repeated_failures_remove_cache() {
1558        let store = FakeStore::default();
1559        let mut e = engine(&store);
1560        // Seed a persisted settle.
1561        e.tick(1_000.0, 12_500_000, Some(40.0), false);
1562        e.tick(2_000.0, 25_000_000, Some(40.0), false);
1563        e.tick(3_000.0, 26_000_000, Some(40.0), false);
1564        e.tick(4_000.0, 27_040_000, Some(40.0), false);
1565        e.transfer_end("https://h.example/", &store, 4_000.0, true);
1566        assert!(store.get(&tune_key("https://h.example/")).is_some());
1567
1568        // Three consecutive failed transfers invalidate the cache.
1569        for _ in 0..3 {
1570            e.transfer_end("https://h.example/", &store, 5_000.0, false);
1571        }
1572        assert!(store.get(&tune_key("https://h.example/")).is_none());
1573    }
1574
1575    #[test]
1576    fn rtt_drift_re_ramps_from_cache() {
1577        let store = FakeStore::default();
1578        {
1579            let mut e = engine(&store);
1580            e.tick(1_000.0, 12_500_000, Some(400.0), false);
1581            e.tick(2_000.0, 25_000_000, Some(400.0), false);
1582            e.tick(3_000.0, 26_000_000, Some(400.0), false);
1583            e.tick(4_000.0, 27_040_000, Some(400.0), false);
1584            e.transfer_end("https://h.example/", &store, 4_000.0, true);
1585        }
1586        let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
1587        e2.set_direction(TransferKind::Upload);
1588        e2.begin_transfer("https://h.example/", &caps(), &store, 10_000.0, &static_params());
1589        assert_eq!(e2.phase(), TunePhase::Settled);
1590        // New session measures RTT 3× the cached 400 ms → >50% drift → re-ramp.
1591        let ev = e2.tick(11_000.0, 12_500_000, Some(1_200.0), false).unwrap();
1592        assert_eq!(ev.phase, TunePhase::Ramping);
1593        assert!(
1594            ev.params.upload_window <= 2,
1595            "re-ramp starts from mins (first measured window may raise once)"
1596        );
1597    }
1598
1599    #[test]
1600    fn upload_auto_level_benchmarks_once_and_caches() {
1601        let store = FakeStore::default();
1602        let mut e = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Auto);
1603        e.begin_transfer("https://h.example/", &caps(), &store, 0.0, &static_params());
1604        let sample = vec![0xABu8; 256 * 1024];
1605        let lvl1 = e.upload_compress_level(&caps(), Some(&sample), 50.0);
1606        let lvl2 = e.upload_compress_level(&caps(), Some(&sample), 50.0);
1607        assert_eq!(lvl1, lvl2, "benchmark runs once, result cached");
1608        assert!(e.auto_level_ready());
1609        assert!((-8..=4).contains(&lvl1), "level inside advertised range, got {lvl1}");
1610        // Incompressible sample → cheapest level.
1611        let mut e2 = TuningEngine::new(true, DEFAULT_TUNE_TTL_MS, CompressLevel::Auto);
1612        e2.begin_transfer("https://h.example/", &caps(), &store, 0.0, &static_params());
1613        let lvl = e2.upload_compress_level(&caps(), Some(&[0u8; 0]), 50.0);
1614        assert_eq!(lvl, caps().compression.zrip_levels.default, "no sample → server default");
1615    }
1616
1617    #[test]
1618    fn disabled_engine_uses_static_params() {
1619        let store = FakeStore::default();
1620        let mut e = TuningEngine::new(false, DEFAULT_TUNE_TTL_MS, CompressLevel::Balanced);
1621        e.set_direction(TransferKind::Upload);
1622        let p = e.begin_transfer("https://h.example/", &caps(), &store, 0.0, &static_params());
1623        assert_eq!(p.concurrency, 4, "static config preserved");
1624        assert_eq!(p.upload_window, 8);
1625        assert_eq!(e.phase(), TunePhase::Uninitialized);
1626        // Ticks are no-ops.
1627        assert!(e.tick(1_000.0, 12_500_000, Some(40.0), false).is_none());
1628        assert_eq!(p, static_params().clamped_into(&caps()));
1629    }
1630}