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