1use std::cell::RefCell;
49use std::rc::Rc;
50
51use libfw_core::{Capabilities, IntRange};
52use serde::{Deserialize, Serialize};
53use wasm_bindgen::JsValue;
54
55pub const TUNE_WINDOW_MS: f64 = 1_000.0;
57pub const GAIN_THRESHOLD: f64 = 0.05;
59pub const RTT_INFLATE_THRESHOLD: f64 = 0.30;
61pub const DEGRADE_FACTOR: f64 = 0.5;
63pub const RTT_EWMA_ALPHA: f64 = 0.25;
65
66pub const HARD_MAX_CONCURRENCY: i64 = 64;
76pub const HARD_MAX_WINDOW: i64 = 64;
78pub const HARD_MAX_CHUNK_SIZE: i64 = 16 * 1024 * 1024;
80pub const HARD_MAX_IN_FLIGHT_BYTES: u64 = 256 * 1024 * 1024;
86pub const HARD_MAX_IN_FLIGHT_REQUESTS: usize = 64;
94pub const LEVEL_SAMPLE_SIZE: usize = 64 * 1024;
103
104#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
110#[serde(rename_all = "lowercase")]
111pub enum TunePhase {
112 Uninitialized,
114 Ramping,
116 Settled,
118 Degraded,
120}
121
122#[derive(Debug, Clone, Copy, PartialEq, Default, Serialize, Deserialize)]
124#[serde(rename_all = "camelCase")]
125pub struct TuneStats {
126 pub rtt_ms: f64,
128 pub mbps: f64,
130}
131
132#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
134#[serde(rename_all = "camelCase")]
135pub struct TuneParams {
136 pub concurrency: usize,
138 pub upload_window: usize,
140 pub download_window: usize,
142 pub chunk_size: u64,
144 pub compress_level: i32,
146}
147
148pub fn request_budget(concurrency: usize, per_file_window: usize) -> usize {
158 concurrency
159 .max(1)
160 .saturating_mul(per_file_window.max(1))
161 .clamp(1, HARD_MAX_IN_FLIGHT_REQUESTS)
162}
163
164pub const DEFAULT_TUNE_TTL_MS: u64 = 60 * 60 * 1_000;
173
174pub const TUNE_CACHE_VERSION: u32 = 1;
177
178pub trait TuneStore {
185 fn get(&self, key: &str) -> Option<String>;
187 fn set(&self, key: &str, value: &str);
189 fn remove(&self, key: &str);
191}
192
193#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
195pub struct TuneCache {
196 pub v: u32,
198 pub caps_hash: String,
200 pub saved_at_ms: u64,
202 pub params: TuneParamsCache,
204}
205
206#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
212pub struct TuneParamsCache {
213 pub concurrency: usize,
214 pub upload_window: usize,
215 pub download_window: usize,
216 pub chunk_size: u64,
217}
218
219impl TuneParamsCache {
220 fn of(params: &TuneParams) -> Self {
221 TuneParamsCache {
222 concurrency: params.concurrency,
223 upload_window: params.upload_window,
224 download_window: params.download_window,
225 chunk_size: params.chunk_size,
226 }
227 }
228
229 fn into_params(self, compress_level: i32) -> TuneParams {
230 TuneParams {
231 concurrency: self.concurrency,
232 upload_window: self.upload_window,
233 download_window: self.download_window,
234 chunk_size: self.chunk_size,
235 compress_level,
236 }
237 }
238}
239
240pub fn tune_key(origin: &str, direction: TransferKind) -> String {
247 let dir = match direction {
248 TransferKind::Upload => "upload",
249 TransferKind::Download => "download",
250 };
251 format!("libfw.tune.v{TUNE_CACHE_VERSION}.{dir}.{origin}")
252}
253
254pub fn origin_of(url: &str) -> String {
259 let (scheme, rest) = match url.split_once("://") {
260 Some((s, r)) => (s, r),
261 None => return String::new(),
262 };
263 if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
264 return String::new();
265 }
266 let authority = rest
268 .split(['/', '?', '#'])
269 .next()
270 .unwrap_or("")
271 .trim();
272 if authority.is_empty() {
273 return String::new();
274 }
275 format!("{}://{}", scheme.to_ascii_lowercase(), authority.to_ascii_lowercase())
276}
277
278impl TuneParams {
279 pub fn request_budget(&self) -> usize {
281 request_budget(self.concurrency, self.upload_window.max(self.download_window))
282 }
283
284 pub fn from_caps_mins(caps: &Capabilities) -> TuneParams {
286 TuneParams {
287 concurrency: floor_of(&caps.limits.concurrency, HARD_MAX_CONCURRENCY) as usize,
288 upload_window: floor_of(&caps.limits.upload_window, HARD_MAX_WINDOW) as usize,
289 download_window: floor_of(&caps.limits.download_window, HARD_MAX_WINDOW) as usize,
290 chunk_size: floor_of(&caps.limits.chunk_size, HARD_MAX_CHUNK_SIZE) as u64,
291 compress_level: caps.compression.zrip_levels.min,
292 }
293 }
294
295 pub fn from_config(
298 concurrency: usize,
299 upload_window: usize,
300 download_window: usize,
301 chunk_size: u64,
302 compress_level: i32,
303 caps: &Capabilities,
304 ) -> TuneParams {
305 TuneParams {
306 concurrency: bounded(
307 concurrency.max(1) as i64,
308 &caps.limits.concurrency,
309 HARD_MAX_CONCURRENCY,
310 ) as usize,
311 upload_window: bounded(
312 upload_window.max(1) as i64,
313 &caps.limits.upload_window,
314 HARD_MAX_WINDOW,
315 ) as usize,
316 download_window: bounded(
317 download_window.max(1) as i64,
318 &caps.limits.download_window,
319 HARD_MAX_WINDOW,
320 ) as usize,
321 chunk_size: bounded(
322 chunk_size.max(1) as i64,
323 &caps.limits.chunk_size,
324 HARD_MAX_CHUNK_SIZE,
325 ) as u64,
326 compress_level: caps.clamp_level(compress_level),
327 }
328 }
329
330 pub fn clamped_into(&self, caps: &Capabilities) -> TuneParams {
333 TuneParams {
334 concurrency: bounded(
335 self.concurrency as i64,
336 &caps.limits.concurrency,
337 HARD_MAX_CONCURRENCY,
338 ) as usize,
339 upload_window: bounded(
340 self.upload_window as i64,
341 &caps.limits.upload_window,
342 HARD_MAX_WINDOW,
343 ) as usize,
344 download_window: bounded(
345 self.download_window as i64,
346 &caps.limits.download_window,
347 HARD_MAX_WINDOW,
348 ) as usize,
349 chunk_size: bounded(
350 self.chunk_size as i64,
351 &caps.limits.chunk_size,
352 HARD_MAX_CHUNK_SIZE,
353 ) as u64,
354 compress_level: caps.clamp_level(self.compress_level),
355 }
356 }
357
358 pub fn upload_in_flight(&self) -> u64 {
360 self.concurrency as u64 * self.upload_window as u64 * self.chunk_size
361 }
362
363 pub fn download_in_flight(&self) -> u64 {
365 self.concurrency as u64 * self.download_window as u64 * self.chunk_size
366 }
367}
368
369fn floor_of(range: &IntRange, hard_max: i64) -> i64 {
375 range.min.clamp(1, hard_max.max(1))
376}
377
378fn bounded(v: i64, range: &IntRange, hard_max: i64) -> i64 {
384 let hard_max = hard_max.max(1);
385 let v = v.max(1);
386 if range.min > range.max {
387 return v.min(hard_max).max(1);
390 }
391 let lo = range.min.clamp(1, hard_max);
392 let hi = range.max.min(hard_max).max(lo);
393 v.clamp(lo, hi)
394}
395
396pub fn ewma(prev: Option<f64>, sample: f64, alpha: f64) -> f64 {
402 match prev {
403 None => sample,
404 Some(p) => alpha * sample + (1.0 - alpha) * p,
405 }
406}
407
408pub fn bdp_bytes(mbps: f64, rtt_ms: f64) -> u64 {
410 (mbps * 1e6 / 8.0 * rtt_ms / 1_000.0).max(0.0) as u64
411}
412
413pub const TARGET_CHUNK_SPAN_MS: f64 = 100.0;
430
431const CHUNK_ALIGN: u64 = 64 * 1024;
438
439pub fn chunk_for_link(mbps: f64, range: &IntRange, budget_max: u64) -> u64 {
445 let floor = floor_of(range, HARD_MAX_CHUNK_SIZE).max(1) as u64;
446 if !mbps.is_finite() || mbps <= 0.0 {
447 return floor.min(budget_max.max(1)).max(1);
448 }
449 let bytes = (mbps * 1e6 / 8.0 * TARGET_CHUNK_SPAN_MS / 1_000.0).clamp(0.0, i64::MAX as f64) as u64;
451 let aligned = (bytes / CHUNK_ALIGN).saturating_mul(CHUNK_ALIGN);
452 let clamped = bounded(aligned.max(1) as i64, range, HARD_MAX_CHUNK_SIZE) as u64;
453 clamped.clamp(1, budget_max.max(1))
454}
455
456#[derive(Debug, Clone, Copy, PartialEq, Eq)]
458pub enum RampDim {
459 Window,
461 Concurrency,
463}
464
465#[derive(Debug, Clone, Copy, PartialEq, Eq)]
467pub enum RampAction {
468 Raise,
470 AdvanceDim,
472 Settle,
474 Degrade,
476 Hold,
478}
479
480#[derive(Debug, Clone, Copy)]
482pub struct RampInput {
483 pub mbps: f64,
485 pub prev_mbps: f64,
487 pub rtt_inflation: f64,
489 pub errors: u32,
491 pub degraded: bool,
493 pub ramping: bool,
498 pub low_gain_windows: u32,
500 pub in_flight: u64,
502 pub bdp: u64,
510 pub at_cap: bool,
512 pub last_dim: bool,
514}
515
516pub fn ramp_action(input: &RampInput) -> (RampAction, u32) {
535 if input.errors > 0 {
536 return (RampAction::Degrade, 0);
537 }
538 if input.at_cap {
539 return if input.last_dim {
540 (RampAction::Settle, 0)
541 } else {
542 (RampAction::AdvanceDim, 0)
543 };
544 }
545 if input.bdp > 0 && input.in_flight >= input.bdp {
546 return (RampAction::Settle, 0);
547 }
548 if input.rtt_inflation > RTT_INFLATE_THRESHOLD {
549 return (RampAction::Settle, 0);
550 }
551 let gain = if input.degraded {
552 0.0
556 } else if input.prev_mbps > 0.0 {
557 input.mbps / input.prev_mbps - 1.0
558 } else if input.ramping {
559 1.0
562 } else {
563 0.0
568 };
569 if input.ramping && gain >= GAIN_THRESHOLD {
574 return (RampAction::Raise, 0);
575 }
576 let low = input.low_gain_windows + 1;
577 if low >= 2 {
578 if input.degraded || input.last_dim {
585 (RampAction::Settle, low)
586 } else {
587 (RampAction::AdvanceDim, 0)
588 }
589 } else {
590 (RampAction::Hold, low)
591 }
592}
593
594#[derive(Debug, Clone, Copy, PartialEq, Eq)]
596pub enum CompressLevel {
597 Auto,
599 Fast,
601 Balanced,
603 Max,
605 Fixed(i32),
607}
608
609pub fn resolve_level(level: CompressLevel, caps: &Capabilities) -> i32 {
613 let z = caps.compression.zrip_levels;
614 match level {
615 CompressLevel::Fast => z.min,
616 CompressLevel::Balanced | CompressLevel::Auto => z.default,
617 CompressLevel::Max => z.max,
618 CompressLevel::Fixed(l) => z.clamp_level(l),
619 }
620}
621
622pub fn choose_auto_level(
631 candidates: &[(i32, usize, f64)],
632 uncompressed: usize,
633 mbps: f64,
634) -> i32 {
635 if candidates.is_empty() {
636 return 0;
637 }
638 let bytes_per_ms = if mbps > 0.0 {
640 mbps * 1e6 / 8.0 / 1_000.0
641 } else {
642 10e6 / 8.0 / 1_000.0
644 };
645 let max_saved = candidates
646 .iter()
647 .map(|&(_, clen, _)| uncompressed.saturating_sub(clen))
648 .max()
649 .unwrap_or(0);
650 if max_saved == 0 {
651 return candidates.iter().map(|&(l, _, _)| l).min().unwrap_or(0);
653 }
654 let mut best = (candidates[0].0, f64::NEG_INFINITY);
655 for &(lvl, clen, cms) in candidates {
656 let saved = uncompressed.saturating_sub(clen) as f64;
657 let score = saved / bytes_per_ms - cms;
658 if score > best.1 {
660 best = (lvl, score);
661 }
662 }
663 best.0
664}
665
666#[derive(Debug, Clone)]
672pub struct TuneEvent {
673 pub phase: TunePhase,
674 pub params: TuneParams,
675 pub stats: TuneStats,
676}
677
678pub type TuneHandle = Rc<RefCell<TuningEngine>>;
680
681pub struct TuningEngine {
683 enabled: bool,
684 level_cfg: CompressLevel,
685 policy_level: i32,
689 phase: TunePhase,
690 caps: Option<Capabilities>,
691 caps_hash: String,
692 params: TuneParams,
693 stats: TuneStats,
694 auto_level: Option<i32>,
696 auto_level_benchmarked: bool,
698 window_start_ms: f64,
700 window_done_base: u64,
701 window_errors: u32,
702 windows_closed: u32,
703 rtt_ewma: Option<f64>,
704 rtt_baseline: Option<f64>,
710 prev_mbps: f64,
711 low_gain_windows: u32,
712 dim: RampDim,
713 degraded_windows: u32,
714 settled: Option<TuneParams>,
718 settled_caps_hash: String,
720 settled_direction: Option<TransferKind>,
723 post_degrade_settle: bool,
726 direction: TransferKind,
728 store: Option<Rc<dyn TuneStore>>,
731 cache_ttl_ms: u64,
733 origin: String,
736 on_tuning: Option<js_sys::Function>,
738}
739
740impl TuningEngine {
741 pub fn new(enabled: bool, level_cfg: CompressLevel) -> TuningEngine {
743 TuningEngine {
744 enabled,
745 level_cfg,
746 policy_level: libfw_core::ZRIP_DEFAULT_LEVEL,
747 phase: TunePhase::Uninitialized,
748 caps: None,
749 caps_hash: String::new(),
750 params: TuneParams {
752 concurrency: 1,
753 upload_window: 1,
754 download_window: 1,
755 chunk_size: 1,
756 compress_level: 0,
757 },
758 stats: TuneStats::default(),
759 auto_level: None,
760 auto_level_benchmarked: false,
761 window_start_ms: 0.0,
762 window_done_base: 0,
763 window_errors: 0,
764 windows_closed: 0,
765 rtt_ewma: None,
766 rtt_baseline: None,
767 prev_mbps: 0.0,
768 low_gain_windows: 0,
769 dim: RampDim::Window,
770 degraded_windows: 0,
771 settled: None,
772 settled_caps_hash: String::new(),
773 settled_direction: None,
774 post_degrade_settle: false,
775 direction: TransferKind::Download,
776 store: None,
777 cache_ttl_ms: DEFAULT_TUNE_TTL_MS,
778 origin: String::new(),
779 on_tuning: None,
780 }
781 }
782
783 pub fn set_cache(&mut self, store: Rc<dyn TuneStore>, ttl_ms: u64) {
789 self.store = Some(store);
790 self.cache_ttl_ms = ttl_ms;
791 }
792
793 pub fn cache_ttl_ms(&self) -> u64 {
795 self.cache_ttl_ms
796 }
797
798 pub fn set_origin(&mut self, url: &str) {
800 self.origin = origin_of(url);
801 }
802
803 fn cache_key(&self) -> Option<String> {
806 if self.store.is_none() || self.cache_ttl_ms == 0 || self.origin.is_empty() {
807 return None;
808 }
809 Some(tune_key(&self.origin, self.direction))
810 }
811
812 fn remove_cache_entry(&self) {
814 if let (Some(store), Some(key)) = (&self.store, self.cache_key()) {
815 store.remove(&key);
816 }
817 }
818
819 fn load_cached_settle(&mut self, now_ms: f64, caps: &Capabilities) -> Option<TuneParams> {
827 let key = self.cache_key()?;
828 let store = self.store.clone()?;
829 let raw = store.get(&key)?;
830 let entry: TuneCache = match serde_json::from_str(&raw) {
831 Ok(entry) => entry,
832 Err(_) => {
833 store.remove(&key);
835 return None;
836 }
837 };
838 let aged = now_ms - entry.saved_at_ms as f64;
839 let fresh = aged >= 0.0 && aged <= self.cache_ttl_ms as f64;
840 if entry.v != TUNE_CACHE_VERSION || entry.caps_hash != caps.caps_hash() || !fresh {
841 store.remove(&key);
842 return None;
843 }
844 Some(entry.params.into_params(self.policy_level))
845 }
846
847 fn save_cached_settle(&self, now_ms: f64) {
849 let Some(key) = self.cache_key() else { return };
850 let Some(store) = self.store.as_ref() else { return };
851 let entry = TuneCache {
852 v: TUNE_CACHE_VERSION,
853 caps_hash: self.caps_hash.clone(),
854 saved_at_ms: if now_ms > 0.0 { now_ms as u64 } else { 0 },
855 params: TuneParamsCache::of(&self.params),
856 };
857 if let Ok(json) = serde_json::to_string(&entry) {
858 store.set(&key, &json);
859 }
860 }
861
862 pub fn set_direction(&mut self, kind: TransferKind) {
865 self.direction = kind;
866 }
867
868 pub fn direction(&self) -> TransferKind {
870 self.direction
871 }
872
873 pub fn set_on_tuning(&mut self, cb: Option<js_sys::Function>) {
875 self.on_tuning = cb;
876 }
877
878 pub fn phase(&self) -> TunePhase {
880 self.phase
881 }
882
883 pub fn params(&self) -> TuneParams {
885 self.params.clone()
886 }
887
888 pub fn stats(&self) -> TuneStats {
890 self.stats
891 }
892
893 pub fn enabled(&self) -> bool {
895 self.enabled
896 }
897
898 pub fn set_enabled(&mut self, enabled: bool) {
901 self.enabled = enabled;
902 }
903
904 pub fn caps(&self) -> Option<Capabilities> {
906 self.caps.clone()
907 }
908
909 pub fn state_js(&self) -> JsValue {
912 let o = js_sys::Object::new();
913 let phase = match self.phase {
914 TunePhase::Uninitialized => "uninitialized",
915 TunePhase::Ramping => "ramping",
916 TunePhase::Settled => "settled",
917 TunePhase::Degraded => "degraded",
918 };
919 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("phase"), &JsValue::from_str(phase));
920 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("params"), ¶ms_to_js(&self.params));
921 let stats = js_sys::Object::new();
922 let _ = js_sys::Reflect::set(
923 &stats,
924 &JsValue::from_str("rttMs"),
925 &JsValue::from_f64(self.stats.rtt_ms),
926 );
927 let _ = js_sys::Reflect::set(
928 &stats,
929 &JsValue::from_str("mbps"),
930 &JsValue::from_f64(self.stats.mbps),
931 );
932 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("stats"), &stats.into());
933 let _ = js_sys::Reflect::set(
934 &o,
935 &JsValue::from_str("capsHash"),
936 &JsValue::from_str(&self.caps_hash),
937 );
938 o.into()
939 }
940
941 pub fn caps_hash(&self) -> &str {
943 &self.caps_hash
944 }
945
946 pub fn begin_transfer(
960 &mut self,
961 caps: &Capabilities,
962 now_ms: f64,
963 static_params: &TuneParams,
964 ) -> TuneParams {
965 self.caps = Some(caps.clone());
966 self.caps_hash = caps.caps_hash();
967 self.windows_closed = 0;
968 self.window_errors = 0;
969 self.window_done_base = 0;
970 self.window_start_ms = now_ms;
971 self.auto_level_benchmarked = false;
972 self.auto_level = None;
973 self.post_degrade_settle = false;
974 self.prev_mbps = 0.0;
979 self.low_gain_windows = 0;
980 self.rtt_ewma = None;
981 self.rtt_baseline = None;
982 self.degraded_windows = 0;
983
984 let policy_level = static_params.compress_level;
988 self.policy_level = policy_level;
989
990 if !self.enabled {
991 self.phase = TunePhase::Uninitialized;
992 self.params = static_params.clamped_into(caps);
993 self.enforce_in_flight_budget();
994 return self.params.clone();
995 }
996
997 let in_memory = self.settled.clone().filter(|_| {
1001 self.settled_caps_hash == self.caps_hash
1002 && self.settled_direction == Some(self.direction)
1003 });
1004 let reusable = match in_memory {
1005 Some(settled) => Some(settled),
1006 None => self.load_cached_settle(now_ms, caps),
1007 };
1008 match reusable {
1009 Some(settled) => {
1010 self.params = settled.clamped_into(caps);
1011 self.params.compress_level = policy_level;
1012 self.phase = TunePhase::Settled;
1013 }
1014 None => {
1015 self.params = TuneParams::from_caps_mins(caps);
1016 self.params.compress_level = policy_level;
1017 self.stats = TuneStats::default();
1018 self.phase = TunePhase::Ramping;
1019 }
1020 }
1021 self.dim = RampDim::Window;
1022 self.enforce_in_flight_budget();
1025 self.params.clone()
1026 }
1027
1028 pub fn tick(
1035 &mut self,
1036 now_ms: f64,
1037 done_bytes: u64,
1038 rtt_ms: Option<f64>,
1039 error: bool,
1040 ) -> Option<TuneEvent> {
1041 if error {
1042 self.window_errors += 1;
1043 }
1044 if let Some(r) = rtt_ms.filter(|&r| r > 0.0) {
1045 let new_ewma = ewma(self.rtt_ewma, r, RTT_EWMA_ALPHA);
1046 if self.enabled && self.rtt_baseline.is_none() {
1047 self.rtt_baseline = Some(new_ewma);
1051 }
1052 self.rtt_ewma = Some(new_ewma);
1053 }
1054 let elapsed = now_ms - self.window_start_ms;
1055 if elapsed < TUNE_WINDOW_MS {
1056 return None;
1057 }
1058 if !self.enabled {
1059 return None;
1060 }
1061 let caps = self.caps.clone()?;
1062
1063 let bytes = done_bytes.saturating_sub(self.window_done_base);
1064 let mbps = if elapsed > 0.0 {
1065 bytes as f64 / elapsed * 1_000.0 * 8.0 / 1e6
1066 } else {
1067 0.0
1068 };
1069 self.stats.mbps = mbps;
1070 if let Some(rtt) = self.rtt_ewma {
1071 self.stats.rtt_ms = rtt;
1072 }
1073
1074 self.retune_chunk_size(mbps, &caps);
1083
1084 let ramping = self.phase == TunePhase::Ramping;
1088 let input = RampInput {
1089 mbps,
1090 prev_mbps: self.prev_mbps,
1091 rtt_inflation: if ramping { self.rtt_inflation() } else { 0.0 },
1092 errors: self.window_errors,
1093 degraded: self.phase == TunePhase::Degraded,
1094 ramping,
1095 low_gain_windows: self.low_gain_windows,
1096 in_flight: self.in_flight(),
1097 bdp: if ramping && self.dim == RampDim::Concurrency {
1100 bdp_bytes(mbps, self.rtt_ewma.unwrap_or(self.stats.rtt_ms))
1101 } else {
1102 0
1103 },
1104 at_cap: self.dim_at_cap(&caps),
1105 last_dim: self.dim == RampDim::Concurrency,
1106 };
1107 let (action, low) = ramp_action(&input);
1108 self.low_gain_windows = low;
1109 self.windows_closed += 1;
1110 self.prev_mbps = mbps;
1111
1112 match action {
1113 RampAction::Raise => {
1114 self.raise_dim(&caps);
1115 self.enforce_in_flight_budget();
1116 if self.phase == TunePhase::Degraded {
1117 self.phase = TunePhase::Ramping;
1118 }
1119 }
1120 RampAction::AdvanceDim => {
1121 self.dim = next_dim(self.dim);
1122 }
1123 RampAction::Settle => {
1124 self.phase = TunePhase::Settled;
1125 self.low_gain_windows = 0;
1126 if !self.post_degrade_settle {
1131 self.settled = Some(self.params.clone());
1132 self.settled_caps_hash = self.caps_hash.clone();
1133 self.settled_direction = Some(self.direction);
1134 self.save_cached_settle(now_ms);
1135 }
1136 }
1137 RampAction::Degrade => {
1138 self.halve_dim();
1139 self.phase = TunePhase::Degraded;
1140 self.degraded_windows = 0;
1141 self.low_gain_windows = 0;
1142 }
1143 RampAction::Hold => {
1144 if self.phase == TunePhase::Degraded {
1145 self.degraded_windows += 1;
1147 if self.degraded_windows >= 2 {
1148 self.phase = TunePhase::Settled;
1149 self.post_degrade_settle = true;
1155 }
1156 }
1157 }
1158 }
1159
1160 self.window_start_ms = now_ms;
1162 self.window_done_base = done_bytes;
1163 self.window_errors = 0;
1164
1165 let event = TuneEvent {
1166 phase: self.phase,
1167 params: self.params.clone(),
1168 stats: self.stats,
1169 };
1170 self.emit(&event);
1171 Some(event)
1172 }
1173
1174 pub fn transfer_end(&mut self, ok: bool) {
1180 if !self.enabled {
1181 return;
1182 }
1183 if !ok {
1184 self.settled = None;
1185 self.settled_caps_hash.clear();
1186 self.settled_direction = None;
1187 self.remove_cache_entry();
1188 self.phase = TunePhase::Ramping;
1189 }
1190 }
1191
1192 pub fn upload_compress_level(
1198 &mut self,
1199 caps: &Capabilities,
1200 sample: Option<&[u8]>,
1201 mbps: f64,
1202 ) -> i32 {
1203 match self.level_cfg {
1204 CompressLevel::Auto => {
1205 if !self.auto_level_benchmarked {
1206 self.auto_level_benchmarked = true;
1207 self.auto_level = Some(match sample {
1208 Some(s) if !s.is_empty() => {
1209 let s = &s[..LEVEL_SAMPLE_SIZE.min(s.len())];
1213 let candidates = auto_candidates(caps);
1214 let results = benchmark_levels(s, &candidates);
1215 choose_auto_level(&results, s.len(), mbps.max(1.0))
1216 }
1217 _ => caps.compression.zrip_levels.default,
1218 });
1219 }
1220 self.auto_level.unwrap_or(caps.compression.zrip_levels.default)
1221 }
1222 other => resolve_level(other, caps),
1223 }
1224 }
1225
1226 pub fn auto_level_ready(&self) -> bool {
1228 self.auto_level_benchmarked
1229 }
1230
1231 fn rtt_inflation(&self) -> f64 {
1232 match (self.rtt_ewma, self.rtt_baseline) {
1236 (Some(cur), Some(base)) if base > 0.0 => (cur / base - 1.0).max(0.0),
1237 _ => 0.0,
1238 }
1239 }
1240
1241 fn in_flight(&self) -> u64 {
1242 match self.last_transfer_kind() {
1243 TransferKind::Upload => self.params.upload_in_flight(),
1244 TransferKind::Download => self.params.download_in_flight(),
1245 }
1246 }
1247
1248 fn dim_value(&self) -> i64 {
1250 match self.dim {
1251 RampDim::Window => match self.last_transfer_kind() {
1252 TransferKind::Upload => self.params.upload_window as i64,
1253 TransferKind::Download => self.params.download_window as i64,
1254 },
1255 RampDim::Concurrency => self.params.concurrency as i64,
1256 }
1257 }
1258
1259 fn dim_max(&self, caps: &Capabilities) -> i64 {
1262 let advertised = match self.dim {
1263 RampDim::Window => match self.last_transfer_kind() {
1264 TransferKind::Upload => caps.limits.upload_window.max,
1265 TransferKind::Download => caps.limits.download_window.max,
1266 },
1267 RampDim::Concurrency => caps.limits.concurrency.max,
1268 };
1269 let hard = match self.dim {
1270 RampDim::Window => HARD_MAX_WINDOW,
1271 RampDim::Concurrency => HARD_MAX_CONCURRENCY,
1272 };
1273 advertised.min(hard).max(1)
1274 }
1275
1276 fn max_chunk_for_budget(&self) -> u64 {
1279 let widest = self.params.upload_window.max(self.params.download_window).max(1) as u64;
1280 let divisor = (self.params.concurrency.max(1) as u64) * widest;
1281 (HARD_MAX_IN_FLIGHT_BYTES / divisor).max(1)
1282 }
1283
1284 fn retune_chunk_size(&mut self, mbps: f64, caps: &Capabilities) {
1291 let target = chunk_for_link(mbps, &caps.limits.chunk_size, self.max_chunk_for_budget());
1292 let current = self.params.chunk_size.max(1);
1293 let grow = current.saturating_mul(3) / 2;
1294 let shrink = (current / 2).max(1);
1295 if target > grow || target < shrink {
1296 self.params.chunk_size = target;
1297 self.enforce_in_flight_budget();
1301 }
1302 }
1303
1304 fn enforce_in_flight_budget(&mut self) {
1310 let max_chunk = self.max_chunk_for_budget();
1311 self.params.chunk_size = self.params.chunk_size.clamp(1, max_chunk);
1312 }
1313
1314 fn dim_at_cap(&self, caps: &Capabilities) -> bool {
1315 self.dim_value() >= self.dim_max(caps)
1316 }
1317
1318 fn raise_dim(&mut self, caps: &Capabilities) {
1319 let dim_max: i64 = self.dim_max(caps);
1320 let step = |v: i64| -> i64 {
1323 if v <= 0 {
1324 1
1325 } else if v * 2 <= dim_max {
1326 v * 2
1327 } else {
1328 (v + 1).min(dim_max)
1329 }
1330 };
1331 match self.dim {
1332 RampDim::Window => match self.last_transfer_kind() {
1333 TransferKind::Upload => {
1334 self.params.upload_window = step(self.params.upload_window as i64).max(1) as usize
1335 }
1336 TransferKind::Download => {
1337 self.params.download_window =
1338 step(self.params.download_window as i64).max(1) as usize
1339 }
1340 },
1341 RampDim::Concurrency => {
1342 self.params.concurrency = step(self.params.concurrency as i64).max(1) as usize
1343 }
1344 }
1345 }
1346
1347 fn halve_dim(&mut self) {
1348 let step = |v: i64| -> i64 { ((v as f64) * DEGRADE_FACTOR).floor().max(1.0) as i64 };
1349 match self.dim {
1350 RampDim::Window => match self.last_transfer_kind() {
1351 TransferKind::Upload => {
1352 self.params.upload_window = step(self.params.upload_window as i64) as usize
1353 }
1354 TransferKind::Download => {
1355 self.params.download_window = step(self.params.download_window as i64) as usize
1356 }
1357 },
1358 RampDim::Concurrency => {
1359 self.params.concurrency = step(self.params.concurrency as i64) as usize
1360 }
1361 }
1362 }
1363}
1364
1365#[derive(Debug, Clone, Copy, PartialEq, Eq)]
1367pub enum TransferKind {
1368 Upload,
1369 Download,
1370}
1371
1372impl TuningEngine {
1373 fn last_transfer_kind(&self) -> TransferKind {
1374 self.direction
1377 }
1378}
1379
1380fn next_dim(dim: RampDim) -> RampDim {
1385 match dim {
1386 RampDim::Window => RampDim::Concurrency,
1387 RampDim::Concurrency => RampDim::Concurrency,
1392 }
1393}
1394
1395fn auto_candidates(caps: &Capabilities) -> Vec<i32> {
1397 let z = caps.compression.zrip_levels;
1398 vec![z.min, z.default, z.max]
1399}
1400
1401fn benchmark_levels(sample: &[u8], levels: &[i32]) -> Vec<(i32, usize, f64)> {
1405 levels
1406 .iter()
1407 .map(|&level| {
1408 let t0 = now_ms();
1409 let out = zrip::compress(sample, level).unwrap_or_else(|_| sample.to_vec());
1410 (level, out.len(), now_ms() - t0)
1411 })
1412 .collect()
1413}
1414
1415pub fn now_ms() -> f64 {
1417 #[cfg(target_arch = "wasm32")]
1418 {
1419 js_sys::Date::now()
1420 }
1421 #[cfg(not(target_arch = "wasm32"))]
1422 {
1423 std::time::SystemTime::now()
1424 .duration_since(std::time::UNIX_EPOCH)
1425 .map(|d| d.as_secs_f64() * 1_000.0)
1426 .unwrap_or(0.0)
1427 }
1428}
1429
1430impl TuningEngine {
1431 fn emit(&self, event: &TuneEvent) {
1432 if let Some(cb) = &self.on_tuning {
1433 let params = params_to_js(&event.params);
1434 let stats = js_sys::Object::new();
1435 let _ = js_sys::Reflect::set(
1436 &stats,
1437 &JsValue::from_str("rttMs"),
1438 &JsValue::from_f64(event.stats.rtt_ms),
1439 );
1440 let _ = js_sys::Reflect::set(
1441 &stats,
1442 &JsValue::from_str("mbps"),
1443 &JsValue::from_f64(event.stats.mbps),
1444 );
1445 let phase = match event.phase {
1446 TunePhase::Uninitialized => "uninitialized",
1447 TunePhase::Ramping => "ramping",
1448 TunePhase::Settled => "settled",
1449 TunePhase::Degraded => "degraded",
1450 };
1451 let _ = cb.call3(
1452 &JsValue::NULL,
1453 &JsValue::from_str(phase),
1454 ¶ms,
1455 &stats.into(),
1456 );
1457 }
1458 }
1459}
1460
1461fn params_to_js(params: &TuneParams) -> JsValue {
1462 let o = js_sys::Object::new();
1463 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("concurrency"), &JsValue::from_f64(params.concurrency as f64));
1464 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("uploadWindow"), &JsValue::from_f64(params.upload_window as f64));
1465 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("downloadWindow"), &JsValue::from_f64(params.download_window as f64));
1466 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("chunkSize"), &JsValue::from_f64(params.chunk_size as f64));
1467 let _ = js_sys::Reflect::set(&o, &JsValue::from_str("compressLevel"), &JsValue::from_f64(params.compress_level as f64));
1468 o.into()
1469}
1470
1471#[cfg(test)]
1472mod tests {
1473 use super::*;
1474
1475 fn caps() -> Capabilities {
1476 Capabilities::default()
1477 }
1478
1479 fn static_params() -> TuneParams {
1480 TuneParams {
1481 concurrency: 4,
1482 upload_window: 8,
1483 download_window: 4,
1484 chunk_size: 2 * 1024 * 1024,
1485 compress_level: 1,
1486 }
1487 }
1488
1489 fn engine() -> TuningEngine {
1491 let mut e = TuningEngine::new(true, CompressLevel::Balanced);
1492 e.set_direction(TransferKind::Upload);
1493 e.begin_transfer(&caps(), 0.0, &static_params());
1494 e
1495 }
1496
1497 fn ramp_to_settle(e: &mut TuningEngine) {
1501 let step = 12_500_000u64;
1502 let mut done = step;
1503 for i in 0..16 {
1504 let t = 1_000.0 * (i as f64 + 1.0);
1505 if let Some(TuneEvent {
1506 phase: TunePhase::Settled,
1507 ..
1508 }) = e.tick(t, done, Some(400.0), false)
1509 {
1510 return;
1511 }
1512 done += step;
1513 }
1514 panic!("engine did not settle");
1515 }
1516
1517 #[test]
1520 fn ramp_action_error_degrades() {
1521 let input = RampInput {
1522 mbps: 10.0,
1523 prev_mbps: 5.0,
1524 rtt_inflation: 0.0,
1525 errors: 1,
1526 degraded: false,
1527 ramping: true,
1528 low_gain_windows: 0,
1529 in_flight: 100,
1530 bdp: 1_000,
1531 at_cap: false,
1532 last_dim: false,
1533 };
1534 assert_eq!(ramp_action(&input).0, RampAction::Degrade);
1535 }
1536
1537 #[test]
1538 fn ramp_action_cap_advances_or_settles() {
1539 let base = RampInput {
1540 mbps: 10.0,
1541 prev_mbps: 5.0,
1542 rtt_inflation: 0.0,
1543 errors: 0,
1544 degraded: false,
1545 ramping: true,
1546 low_gain_windows: 0,
1547 in_flight: 100,
1548 bdp: 1_000,
1549 at_cap: true,
1550 last_dim: false,
1551 };
1552 assert_eq!(ramp_action(&base).0, RampAction::AdvanceDim);
1553 let last = RampInput { last_dim: true, ..base };
1554 assert_eq!(ramp_action(&last).0, RampAction::Settle);
1555 }
1556
1557 #[test]
1558 fn ramp_action_bdp_and_rtt_saturate() {
1559 let base = RampInput {
1560 mbps: 10.0,
1561 prev_mbps: 5.0,
1562 rtt_inflation: 0.0,
1563 errors: 0,
1564 degraded: false,
1565 ramping: true,
1566 low_gain_windows: 0,
1567 in_flight: 2_000,
1568 bdp: 1_000,
1569 at_cap: false,
1570 last_dim: false,
1571 };
1572 assert_eq!(ramp_action(&base).0, RampAction::Settle, "in-flight ≥ BDP settles");
1573
1574 let rtt = RampInput {
1575 rtt_inflation: 0.5,
1576 in_flight: 10,
1577 bdp: 1_000,
1578 ..base
1579 };
1580 assert_eq!(ramp_action(&rtt).0, RampAction::Settle, "RTT inflation settles");
1581 }
1582
1583 #[test]
1584 fn ramp_action_gain_raises_and_low_gain_settles() {
1585 let growing = RampInput {
1587 mbps: 20.0,
1588 prev_mbps: 10.0,
1589 ..RampInput {
1590 mbps: 0.0,
1591 prev_mbps: 0.0,
1592 rtt_inflation: 0.0,
1593 errors: 0,
1594 degraded: false,
1595 ramping: true,
1596 low_gain_windows: 7,
1597 in_flight: 10,
1598 bdp: 1_000,
1599 at_cap: false,
1600 last_dim: false,
1601 }
1602 };
1603 let (action, low) = ramp_action(&growing);
1604 assert_eq!(action, RampAction::Raise);
1605 assert_eq!(low, 0, "raise resets the low-gain counter");
1606
1607 let flat = RampInput {
1609 mbps: 10.4,
1610 prev_mbps: 10.0,
1611 low_gain_windows: 0,
1612 ..growing
1613 };
1614 let (action, low) = ramp_action(&flat);
1615 assert_eq!(action, RampAction::Hold);
1616 assert_eq!(low, 1, "first below-threshold window holds");
1617
1618 let flat2 = RampInput { low_gain_windows: 1, ..flat };
1622 let (action, low) = ramp_action(&flat2);
1623 assert_eq!(action, RampAction::AdvanceDim);
1624 assert_eq!(low, 0, "advancing resets the low-gain counter");
1625
1626 let flat_last = RampInput {
1628 last_dim: true,
1629 low_gain_windows: 1,
1630 ..flat
1631 };
1632 assert_eq!(ramp_action(&flat_last).0, RampAction::Settle);
1633
1634 let first = RampInput {
1636 mbps: 1.0,
1637 prev_mbps: 0.0,
1638 low_gain_windows: 0,
1639 ..growing
1640 };
1641 assert_eq!(ramp_action(&first).0, RampAction::Raise);
1642 }
1643
1644 #[test]
1645 fn ramp_action_degraded_hold_never_raises() {
1646 let degraded = RampInput {
1650 mbps: 0.008,
1651 prev_mbps: 0.0,
1652 degraded: true,
1653 ..RampInput {
1654 mbps: 0.0,
1655 prev_mbps: 0.0,
1656 rtt_inflation: 0.0,
1657 errors: 0,
1658 degraded: false,
1659 ramping: true,
1660 low_gain_windows: 0,
1661 in_flight: 10,
1662 bdp: 1_000,
1663 at_cap: false,
1664 last_dim: false,
1665 }
1666 };
1667 let (action, low) = ramp_action(°raded);
1668 assert_eq!(action, RampAction::Hold);
1669 assert_eq!(low, 1, "first stability window holds");
1670 let (action, low) = ramp_action(&RampInput { low_gain_windows: 1, ..degraded });
1671 assert_eq!(action, RampAction::Settle);
1672 assert_eq!(low, 2, "second stability window settles");
1673 }
1674
1675 #[test]
1676 fn ewma_seeds_and_smooths() {
1677 assert_eq!(ewma(None, 10.0, 0.25), 10.0);
1678 let v = ewma(Some(10.0), 20.0, 0.25);
1679 assert!((v - 12.5).abs() < 1e-9);
1680 }
1681
1682 #[test]
1683 fn bdp_conversion() {
1684 assert_eq!(bdp_bytes(10.0, 100.0), 125_000);
1686 assert_eq!(bdp_bytes(0.0, 0.0), 0);
1687 }
1688
1689 #[test]
1690 fn request_budget_covers_files_and_the_per_file_window() {
1691 assert_eq!(request_budget(1, 1), 1);
1694 assert_eq!(request_budget(1, 8), 8);
1695 assert_eq!(request_budget(4, 8), 32);
1696 assert_eq!(request_budget(0, 0), 1);
1698 assert_eq!(request_budget(64, 64), HARD_MAX_IN_FLIGHT_REQUESTS);
1700
1701 let mut p = static_params();
1703 p.concurrency = 2;
1704 p.upload_window = 8;
1705 p.download_window = 4;
1706 assert_eq!(p.request_budget(), 16);
1707 p.upload_window = 4;
1708 assert_eq!(p.request_budget(), 8);
1709 }
1710
1711 #[test]
1712 fn resolve_level_mapping() {
1713 let c = caps();
1714 assert_eq!(resolve_level(CompressLevel::Fast, &c), -8);
1715 assert_eq!(resolve_level(CompressLevel::Balanced, &c), 1);
1716 assert_eq!(resolve_level(CompressLevel::Max, &c), 4);
1717 assert_eq!(resolve_level(CompressLevel::Auto, &c), 1);
1718 assert_eq!(resolve_level(CompressLevel::Fixed(2), &c), 2);
1719 assert_eq!(resolve_level(CompressLevel::Fixed(99), &c), 4);
1720 assert_eq!(resolve_level(CompressLevel::Fixed(-99), &c), -8);
1721 }
1722
1723 #[test]
1724 fn choose_auto_level_picks_by_net_gain() {
1725 assert_eq!(
1727 choose_auto_level(&[(-8, 1000, 0.1), (1, 1000, 5.0), (4, 1000, 40.0)], 1000, 10.0),
1728 -8
1729 );
1730
1731 let cpu_bound = choose_auto_level(&[(-8, 800_000, 0.1), (4, 400_000, 30.0)], 1_000_000, 1000.0);
1733 assert_eq!(cpu_bound, -8);
1734
1735 let savings = choose_auto_level(&[(-8, 800_000, 0.1), (4, 400_000, 30.0)], 1_000_000, 10.0);
1737 assert_eq!(savings, 4);
1738
1739 assert_eq!(choose_auto_level(&[(0, 500, 1.0), (4, 500, 1.0)], 1000, 100.0), 0);
1741 }
1742
1743 #[test]
1746 fn begin_transfer_miss_ramps_from_mins() {
1747 let mut e = TuningEngine::new(true, CompressLevel::Balanced);
1748 e.set_direction(TransferKind::Upload);
1749 let p = e.begin_transfer(&caps(), 0.0, &static_params());
1750 assert_eq!(e.phase(), TunePhase::Ramping);
1751 assert_eq!(p.concurrency, 1);
1752 assert_eq!(p.upload_window, 1);
1753 assert_eq!(p.chunk_size, 262_144);
1754 }
1755
1756 #[test]
1757 fn settled_reuse_first_window_does_not_raise() {
1758 let input = RampInput {
1762 mbps: 100.0,
1763 prev_mbps: 0.0,
1764 rtt_inflation: 0.0,
1765 errors: 0,
1766 degraded: false,
1767 ramping: false,
1768 low_gain_windows: 0,
1769 in_flight: 10,
1770 bdp: 0,
1771 at_cap: false,
1772 last_dim: false,
1773 };
1774 let (action, low) = ramp_action(&input);
1775 assert_eq!(action, RampAction::Hold);
1776 assert_eq!(low, 1);
1777 }
1778
1779 #[test]
1780 fn reused_params_do_not_raise_on_the_first_window() {
1781 let mut e = engine();
1782 ramp_to_settle(&mut e);
1783 let tuned = e.params().clone();
1784 e.transfer_end(true);
1785 e.begin_transfer(&caps(), 60_000.0, &static_params());
1786 let ev = e.tick(61_000.0, 100_000_000, Some(400.0), false).unwrap();
1789 assert_eq!(ev.phase, TunePhase::Settled);
1790 assert_eq!(
1791 ev.params.upload_window, tuned.upload_window,
1792 "settled reuse must not raise the window"
1793 );
1794 assert_eq!(ev.params.concurrency, tuned.concurrency);
1795 }
1796
1797 #[test]
1798 fn rtt_inflation_settles_the_ramp() {
1799 let mut e = engine();
1800 let ev = e.tick(1_000.0, 12_500_000, Some(100.0), false).unwrap();
1802 assert_eq!(ev.phase, TunePhase::Ramping);
1803 let ev = e.tick(2_000.0, 25_000_000, Some(400.0), false).unwrap();
1805 assert_eq!(ev.phase, TunePhase::Settled, "RTT inflation must settle");
1806 }
1807
1808 #[test]
1809 fn ramp_start_keeps_the_configured_compress_policy() {
1810 let mut e = TuningEngine::new(true, CompressLevel::Balanced);
1811 e.set_direction(TransferKind::Upload);
1812 let sp = static_params(); let p = e.begin_transfer(&caps(), 0.0, &sp);
1814 assert_eq!(
1815 p.compress_level, 1,
1816 "a ramp must not silently drop to the advertised minimum (-8)"
1817 );
1818 ramp_to_settle(&mut e);
1821 e.transfer_end(true);
1822 let p2 = e.begin_transfer(&caps(), 60_000.0, &sp);
1823 assert_eq!(p2.compress_level, 1, "reuse keeps the policy level");
1824 }
1825
1826 #[test]
1827 fn absurd_advertised_caps_are_clamped_to_hard_ceilings() {
1828 let mut hostile = caps();
1829 hostile.limits.concurrency = IntRange {
1830 default: 1,
1831 max: 100_000,
1832 min: 1_000,
1833 };
1834 hostile.limits.upload_window = IntRange {
1835 default: 1,
1836 max: 100_000,
1837 min: 500,
1838 };
1839 hostile.limits.download_window = IntRange {
1840 default: 1,
1841 max: 100_000,
1842 min: 500,
1843 };
1844 hostile.limits.chunk_size = IntRange {
1845 default: i64::MAX,
1846 max: i64::MAX,
1847 min: i64::MAX,
1848 };
1849 let mut e = TuningEngine::new(true, CompressLevel::Balanced);
1850 e.set_direction(TransferKind::Upload);
1851 let p = e.begin_transfer(&hostile, 0.0, &static_params());
1852 assert!(p.concurrency <= HARD_MAX_CONCURRENCY as usize);
1853 assert!(p.upload_window <= HARD_MAX_WINDOW as usize);
1854 assert!(p.download_window <= HARD_MAX_WINDOW as usize);
1855 assert!(p.chunk_size <= HARD_MAX_CHUNK_SIZE as u64);
1856 assert!(
1857 p.upload_in_flight() <= HARD_MAX_IN_FLIGHT_BYTES,
1858 "in-flight bytes {} exceed the budget",
1859 p.upload_in_flight()
1860 );
1861 }
1862
1863 #[test]
1864 fn inverted_advertised_range_does_not_panic() {
1865 let mut broken = caps();
1866 broken.limits.concurrency = IntRange {
1867 default: 4,
1868 max: 2,
1869 min: 8,
1870 };
1871 broken.limits.chunk_size = IntRange {
1872 default: 0,
1873 max: 0,
1874 min: 0,
1875 };
1876 let mut e = TuningEngine::new(false, CompressLevel::Balanced);
1878 e.set_direction(TransferKind::Upload);
1879 let p = e.begin_transfer(&broken, 0.0, &static_params());
1880 assert!(p.concurrency >= 1 && p.chunk_size >= 1);
1881 let mut e2 = TuningEngine::new(true, CompressLevel::Balanced);
1882 e2.set_direction(TransferKind::Upload);
1883 let p2 = e2.begin_transfer(&broken, 0.0, &static_params());
1884 assert!(p2.concurrency >= 1 && p2.chunk_size >= 1);
1885 }
1886
1887 #[test]
1888 fn ramp_raises_window_then_concurrency_until_cap() {
1889 let mut e = engine();
1890 let mut t = 1_000.0;
1894 let mut done = 12_500_000u64;
1895 let mut delta = 12_500_000u64;
1896 for _ in 0..3 {
1897 let ev = e.tick(t, done, Some(400.0), false).expect("window closed");
1898 assert_eq!(ev.phase, TunePhase::Ramping);
1899 t += 1_000.0;
1900 delta *= 2;
1901 done += delta; }
1903 assert_eq!(e.params().upload_window, 8);
1905 let ev = e.tick(t, done, Some(400.0), false).unwrap();
1906 assert_eq!(ev.params.upload_window, 8); t += 1_000.0;
1908 delta *= 2;
1909 done += delta;
1910 let ev = e.tick(t, done, Some(400.0), false).unwrap();
1912 assert_eq!(ev.params.concurrency, 2);
1913 let ev = e.tick(t + 1_000.0, done + delta * 2, Some(400.0), false).unwrap();
1915 assert_eq!(ev.params.concurrency, 4);
1916 assert_eq!(ev.params.upload_window, 8);
1917 }
1918
1919 #[test]
1920 fn error_window_degrades_and_halves() {
1921 let mut e = engine();
1922 let mut t = 1_000.0;
1924 let mut done = 12_500_000u64;
1925 let mut delta = 12_500_000u64;
1926 for _ in 0..2 {
1927 e.tick(t, done, Some(400.0), false);
1928 t += 1_000.0;
1929 delta *= 2;
1930 done += delta; }
1932 assert_eq!(e.params().upload_window, 4);
1933 let ev = e.tick(t, done, Some(400.0), true).unwrap();
1935 assert_eq!(ev.phase, TunePhase::Degraded);
1936 assert_eq!(ev.params.upload_window, 2);
1937 let ev = e.tick(t + 1_000.0, done + 1_000, None, false).unwrap();
1940 assert_eq!(ev.phase, TunePhase::Degraded);
1941 let ev = e.tick(t + 2_000.0, done + 2_000, None, false).unwrap();
1942 assert_eq!(ev.phase, TunePhase::Settled);
1943 assert_eq!(ev.params.upload_window, 2);
1944 }
1945
1946 #[test]
1947 fn flat_gain_explores_dimensions_then_settles() {
1948 let mut e = engine();
1949 let step = 12_500_000u64;
1954 let mut done = step;
1955 let mut t = 1_000.0;
1956
1957 let ev = e.tick(t, done, Some(400.0), false).unwrap();
1959 assert_eq!(ev.params.upload_window, 2);
1960 for _ in 0..2 {
1961 t += 1_000.0;
1962 done += step;
1963 e.tick(t, done, Some(400.0), false);
1964 }
1965 assert_eq!(
1966 e.phase(),
1967 TunePhase::Ramping,
1968 "a flat window dimension must not end the ramp"
1969 );
1970
1971 t += 1_000.0;
1973 done += step;
1974 let ev = e.tick(t, done, Some(400.0), false).unwrap();
1975 assert_eq!(ev.phase, TunePhase::Ramping);
1976 t += 1_000.0;
1977 done += step;
1978 let ev = e.tick(t, done, Some(400.0), false).unwrap();
1979 assert_eq!(ev.phase, TunePhase::Settled);
1980 assert_eq!(ev.params.upload_window, 2);
1981 e.transfer_end(true);
1982 }
1983
1984 #[test]
1985 fn chunk_size_tracks_the_link_and_ignores_jitter() {
1986 let mut e = engine();
1987 let mut done = 12_500_000u64;
1989 let ev = e.tick(1_000.0, done, Some(400.0), false).unwrap();
1990 let first = ev.params.chunk_size;
1991 assert_eq!(first, 1_245_184, "the chunk must follow the measured link");
1992
1993 let mut t = 1_000.0;
1996 for _ in 0..3 {
1997 t += 1_000.0;
1998 done += 13_500_000; let ev = e.tick(t, done, Some(400.0), false).unwrap();
2000 assert_eq!(ev.params.chunk_size, first, "jitter must not re-size the chunk");
2001 }
2002
2003 t += 1_000.0;
2005 done += 100_000_000; let ev = e.tick(t, done, Some(400.0), false).unwrap();
2007 assert!(ev.params.chunk_size > first, "the chunk must grow on a wider band");
2008 }
2009
2010 #[test]
2011 fn chunk_size_follows_the_measured_link() {
2012 let range = IntRange {
2013 min: 262_144,
2014 default: 2 * 1024 * 1024,
2015 max: 8 * 1024 * 1024,
2016 };
2017 let budget = 8 * 1024 * 1024;
2018 assert_eq!(chunk_for_link(0.0, &range, budget), 262_144);
2020 assert_eq!(chunk_for_link(10.0, &range, budget), 262_144);
2022 assert_eq!(chunk_for_link(30.0, &range, budget), 327_680);
2024 assert_eq!(chunk_for_link(100.0, &range, budget), 1_245_184);
2025 assert_eq!(chunk_for_link(1_000.0, &range, budget), 8 * 1024 * 1024);
2027 assert_eq!(chunk_for_link(1_000.0, &range, 1_500_000), 1_500_000);
2029 let mut last = 0;
2031 for mbps in [1.0, 5.0, 20.0, 80.0, 320.0] {
2032 let v = chunk_for_link(mbps, &range, budget);
2033 assert!(v >= last && v >= 262_144, "{mbps} Mb/s → {v}");
2034 last = v;
2035 }
2036 assert!(chunk_for_link(f64::NAN, &range, budget) >= 1);
2038 assert!(chunk_for_link(-5.0, &range, budget) >= 1);
2039 let inverted = IntRange {
2040 min: 9_000_000,
2041 max: 1,
2042 default: 0,
2043 };
2044 let v = chunk_for_link(50.0, &inverted, budget);
2045 assert!(v >= 1 && v <= budget, "{v}");
2046 }
2047
2048 #[test]
2049 fn settled_state_is_reused_within_the_same_client() {
2050 let mut e = engine();
2051 ramp_to_settle(&mut e);
2052 assert_eq!(e.phase(), TunePhase::Settled);
2053 let tuned = e.params().clone();
2054 e.transfer_end(true);
2055 let p = e.begin_transfer(&caps(), 60_000.0, &static_params());
2058 assert_eq!(e.phase(), TunePhase::Settled, "in-session reuse skips the ramp");
2059 assert_eq!(p.upload_window, tuned.upload_window);
2060 assert_eq!(p.concurrency, tuned.concurrency);
2061 assert_eq!(p.chunk_size, tuned.chunk_size);
2062 assert_eq!(p.compress_level, 1, "reuse keeps the configured policy level");
2063 }
2064
2065 #[test]
2066 fn a_new_engine_always_re_ramps_from_mins() {
2067 let mut e = engine();
2071 ramp_to_settle(&mut e);
2072 e.transfer_end(true);
2073 let fresh = engine();
2074 assert_eq!(fresh.phase(), TunePhase::Ramping);
2075 assert_eq!(fresh.params().upload_window, 1);
2076 assert_eq!(fresh.params().chunk_size, 262_144);
2077 }
2078
2079 #[derive(Clone, Default)]
2083 struct FakeStore(Rc<RefCell<std::collections::HashMap<String, String>>>);
2084
2085 impl FakeStore {
2086 fn len(&self) -> usize {
2087 self.0.borrow().len()
2088 }
2089 fn put_raw(&self, key: &str, value: &str) {
2090 self.0.borrow_mut().insert(key.to_string(), value.to_string());
2091 }
2092 fn keys(&self) -> Vec<String> {
2093 self.0.borrow().keys().cloned().collect()
2094 }
2095 }
2096
2097 impl TuneStore for FakeStore {
2098 fn get(&self, key: &str) -> Option<String> {
2099 self.0.borrow().get(key).cloned()
2100 }
2101 fn set(&self, key: &str, value: &str) {
2102 self.0.borrow_mut().insert(key.to_string(), value.to_string());
2103 }
2104 fn remove(&self, key: &str) {
2105 self.0.borrow_mut().remove(key);
2106 }
2107 }
2108
2109 const ORIGIN: &str = "http://localhost:8080";
2110
2111 fn wired(
2116 store: &FakeStore,
2117 ttl_ms: u64,
2118 direction: TransferKind,
2119 origin: &str,
2120 ) -> TuningEngine {
2121 let mut e = TuningEngine::new(true, CompressLevel::Balanced);
2122 e.set_cache(Rc::new(store.clone()), ttl_ms);
2123 e.set_direction(direction);
2124 e.set_origin(origin);
2125 e
2126 }
2127
2128 fn cached_engine(store: &FakeStore, ttl_ms: u64) -> TuningEngine {
2130 let mut e = wired(store, ttl_ms, TransferKind::Upload, ORIGIN);
2131 e.begin_transfer(&caps(), 0.0, &static_params());
2132 e
2133 }
2134
2135 #[test]
2136 fn a_settle_is_cached_and_reused_by_a_new_engine() {
2137 let store = FakeStore::default();
2138 let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
2139 ramp_to_settle(&mut a);
2140 assert_eq!(a.phase(), TunePhase::Settled);
2141 let tuned = a.params().clone();
2142 a.transfer_end(true);
2143 assert_eq!(store.len(), 1, "the settle is written on convergence");
2144 assert_eq!(
2145 store.keys(),
2146 vec![tune_key(ORIGIN, TransferKind::Upload)],
2147 "keyed by origin + direction"
2148 );
2149
2150 let mut b = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
2152 let p = b.begin_transfer(&caps(), 20_000.0, &static_params());
2153 assert_eq!(b.phase(), TunePhase::Settled, "a cache hit skips the ramp");
2154 assert_eq!(p.upload_window, tuned.upload_window);
2155 assert_eq!(p.concurrency, tuned.concurrency);
2156 assert_eq!(p.chunk_size, tuned.chunk_size);
2157 assert_eq!(p.compress_level, 1, "the policy level is not cached");
2158 }
2159
2160 #[test]
2161 fn a_cached_settle_expires_after_the_ttl() {
2162 let store = FakeStore::default();
2163 let ttl = 60_000;
2164 let mut a = cached_engine(&store, ttl);
2165 ramp_to_settle(&mut a);
2166 let saved = a.params().clone();
2167 assert!(saved.upload_window > 1, "the ramp must have raised something");
2168 a.transfer_end(true);
2169
2170 let row: TuneCache = serde_json::from_str(
2172 &store.get(&tune_key(ORIGIN, TransferKind::Upload)).unwrap(),
2173 )
2174 .unwrap();
2175 let saved_at = row.saved_at_ms as f64;
2176
2177 let mut b = wired(&store, ttl, TransferKind::Upload, ORIGIN);
2179 b.begin_transfer(&caps(), saved_at + ttl as f64 - 1.0, &static_params());
2180 assert_eq!(b.phase(), TunePhase::Settled);
2181
2182 let mut c = wired(&store, ttl, TransferKind::Upload, ORIGIN);
2184 let p = c.begin_transfer(&caps(), saved_at + ttl as f64 + 1.0, &static_params());
2185 assert_eq!(c.phase(), TunePhase::Ramping);
2186 assert_eq!(p.upload_window, 1, "an expired result re-ramps from the mins");
2187 assert_eq!(store.len(), 0, "an expired row is cleaned up");
2188 }
2189
2190 #[test]
2191 fn a_zero_ttl_disables_the_cache() {
2192 let store = FakeStore::default();
2193 assert_eq!(cached_engine(&store, 0).cache_ttl_ms(), 0);
2194 let mut a = cached_engine(&store, 0);
2195 ramp_to_settle(&mut a);
2196 a.transfer_end(true);
2197 assert_eq!(store.len(), 0, "ttl = 0 never writes");
2198
2199 let store2 = FakeStore::default();
2201 let mut x = cached_engine(&store2, DEFAULT_TUNE_TTL_MS);
2202 ramp_to_settle(&mut x);
2203 x.transfer_end(true);
2204 assert_eq!(store2.len(), 1);
2205 let mut y = wired(&store2, 0, TransferKind::Upload, ORIGIN);
2206 y.begin_transfer(&caps(), 20_000.0, &static_params());
2207 assert_eq!(y.phase(), TunePhase::Ramping, "ttl = 0 never reads");
2208 }
2209
2210 #[test]
2211 fn a_cached_settle_is_scoped_to_one_direction() {
2212 let store = FakeStore::default();
2213 let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
2214 ramp_to_settle(&mut a);
2215 a.transfer_end(true);
2216
2217 let mut b = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Download, ORIGIN);
2219 let p = b.begin_transfer(&caps(), 20_000.0, &static_params());
2220 assert_eq!(b.phase(), TunePhase::Ramping);
2221 assert_eq!(p.download_window, 1);
2222
2223 let mut c = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
2225 c.begin_transfer(&caps(), 20_000.0, &static_params());
2226 assert_eq!(c.phase(), TunePhase::Settled);
2227 }
2228
2229 #[test]
2230 fn a_cached_settle_is_scoped_to_one_origin() {
2231 let store = FakeStore::default();
2232 let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
2233 ramp_to_settle(&mut a);
2234 a.transfer_end(true);
2235
2236 let mut b = wired(
2237 &store,
2238 DEFAULT_TUNE_TTL_MS,
2239 TransferKind::Upload,
2240 "https://other.example.com:8443",
2241 );
2242 b.begin_transfer(&caps(), 20_000.0, &static_params());
2243 assert_eq!(b.phase(), TunePhase::Ramping, "another origin is another link");
2244 assert_eq!(store.len(), 1, "the other origin's row is left alone");
2245 }
2246
2247 #[test]
2248 fn an_unusable_origin_disables_the_cache() {
2249 let store = FakeStore::default();
2252 for url in ["", "not a url", "ftp://host/x", "http://", "ws://h:80"] {
2253 let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, url);
2254 let _ = e.begin_transfer(&caps(), 0.0, &static_params());
2255 ramp_to_settle(&mut e);
2256 e.transfer_end(true);
2257 assert_eq!(store.len(), 0, "{url:?} must not be cached");
2258 }
2259 }
2260
2261 #[test]
2262 fn changed_caps_invalidate_a_cached_settle() {
2263 let store = FakeStore::default();
2264 let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
2265 ramp_to_settle(&mut a);
2266 a.transfer_end(true);
2267 assert_eq!(store.len(), 1);
2268
2269 let mut changed = caps();
2270 changed.limits.concurrency.default = 6;
2271 let mut b = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
2272 let p = b.begin_transfer(&changed, 20_000.0, &static_params());
2273 assert_eq!(b.phase(), TunePhase::Ramping);
2274 assert_eq!(p.upload_window, 1);
2275 assert_eq!(store.len(), 0, "a caps mismatch drops the entry");
2276 }
2277
2278 #[test]
2279 fn a_failed_transfer_removes_the_cached_settle() {
2280 let store = FakeStore::default();
2281 let mut a = cached_engine(&store, DEFAULT_TUNE_TTL_MS);
2282 ramp_to_settle(&mut a);
2283 a.transfer_end(false);
2284 assert_eq!(store.len(), 0, "parameters that just failed are not kept");
2285
2286 let p = a.begin_transfer(&caps(), 20_000.0, &static_params());
2288 assert_eq!(a.phase(), TunePhase::Ramping);
2289 assert_eq!(p.upload_window, 1);
2290 }
2291
2292 #[test]
2293 fn a_corrupt_or_foreign_cache_entry_is_dropped() {
2294 let store = FakeStore::default();
2295 let key = tune_key(ORIGIN, TransferKind::Upload);
2296
2297 store.put_raw(&key, "{not json");
2299 let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
2300 e.begin_transfer(&caps(), 20_000.0, &static_params());
2301 assert_eq!(e.phase(), TunePhase::Ramping);
2302 assert_eq!(store.len(), 0, "an unreadable row is removed");
2303
2304 let future = TuneCache {
2306 v: TUNE_CACHE_VERSION + 1,
2307 caps_hash: caps().caps_hash(),
2308 saved_at_ms: 1_000,
2309 params: TuneParamsCache {
2310 concurrency: 12,
2311 upload_window: 16,
2312 download_window: 16,
2313 chunk_size: 4 * 1024 * 1024,
2314 },
2315 };
2316 store.put_raw(&key, &serde_json::to_string(&future).unwrap());
2317 let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
2318 e.begin_transfer(&caps(), 20_000.0, &static_params());
2319 assert_eq!(e.phase(), TunePhase::Ramping);
2320 assert_eq!(store.len(), 0);
2321 }
2322
2323 #[test]
2324 fn a_valid_row_written_by_another_client_is_reused() {
2325 let store = FakeStore::default();
2328 let row = TuneCache {
2329 v: TUNE_CACHE_VERSION,
2330 caps_hash: caps().caps_hash(),
2331 saved_at_ms: 5_000,
2332 params: TuneParamsCache {
2333 concurrency: 3,
2334 upload_window: 7,
2335 download_window: 3,
2336 chunk_size: 1_048_576,
2337 },
2338 };
2339 store.put_raw(
2340 &tune_key(ORIGIN, TransferKind::Upload),
2341 &serde_json::to_string(&row).unwrap(),
2342 );
2343 let mut e = wired(&store, DEFAULT_TUNE_TTL_MS, TransferKind::Upload, ORIGIN);
2344 let p = e.begin_transfer(&caps(), 10_000.0, &static_params());
2345 assert_eq!(e.phase(), TunePhase::Settled);
2346 assert_eq!(p.concurrency, 3);
2347 assert_eq!(p.upload_window, 7);
2348 assert_eq!(p.chunk_size, 1_048_576);
2349 assert_eq!(store.len(), 1, "a fresh, valid row is kept");
2350 }
2351
2352 #[test]
2353 fn tune_key_and_origin_of_normalise_the_scope() {
2354 assert_eq!(
2355 tune_key("http://localhost:8080", TransferKind::Upload),
2356 "libfw.tune.v1.upload.http://localhost:8080"
2357 );
2358 assert_eq!(
2359 tune_key("http://localhost:8080", TransferKind::Download),
2360 "libfw.tune.v1.download.http://localhost:8080"
2361 );
2362 assert_ne!(
2363 tune_key("http://a", TransferKind::Upload),
2364 tune_key("http://a", TransferKind::Download)
2365 );
2366
2367 assert_eq!(origin_of("http://Host:8080/x?y#z"), "http://host:8080");
2368 assert_eq!(origin_of("HTTPS://h/p"), "https://h");
2369 assert_eq!(origin_of("http://h"), "http://h");
2370 assert_eq!(origin_of(""), "");
2372 assert_eq!(origin_of("/relative/path"), "");
2373 assert_eq!(origin_of("ftp://h/x"), "");
2374 assert_eq!(origin_of("http://"), "");
2375 }
2376
2377 #[test]
2378 fn failed_transfer_forgets_the_settled_state() {
2379 let mut e = engine();
2380 ramp_to_settle(&mut e);
2381 e.transfer_end(false);
2382 let p = e.begin_transfer(&caps(), 60_000.0, &static_params());
2384 assert_eq!(e.phase(), TunePhase::Ramping);
2385 assert_eq!(p.upload_window, 1, "a failed transfer re-ramps from the mins");
2386 }
2387
2388 #[test]
2389 fn changed_caps_invalidate_the_settled_state() {
2390 let mut e = engine();
2391 ramp_to_settle(&mut e);
2392 e.transfer_end(true);
2393 let mut changed = caps();
2395 changed.limits.concurrency.default = 6;
2396 let p = e.begin_transfer(&changed, 60_000.0, &static_params());
2397 assert_eq!(e.phase(), TunePhase::Ramping);
2398 assert_eq!(p.upload_window, 1);
2399 }
2400
2401 #[test]
2402 fn reused_settle_does_not_re_ramp_on_later_windows() {
2403 let mut e = engine();
2404 ramp_to_settle(&mut e);
2405 let tuned = e.params().clone();
2406 e.transfer_end(true);
2407 e.begin_transfer(&caps(), 60_000.0, &static_params());
2408 let mut t = 61_000.0;
2412 for i in 1..=3u64 {
2413 let ev = e.tick(t, i * 100_000_000, Some(400.0), false).unwrap();
2414 assert_eq!(ev.phase, TunePhase::Settled);
2415 assert_eq!(ev.params.upload_window, tuned.upload_window);
2416 assert_eq!(ev.params.concurrency, tuned.concurrency);
2417 t += 1_000.0;
2418 }
2419 }
2420
2421 #[test]
2422 fn transfer_end_without_settle_keeps_ramping_next_time() {
2423 let mut e = engine();
2424 e.tick(1_000.0, 12_500_000, Some(400.0), false);
2425 e.transfer_end(true);
2426 let p = e.begin_transfer(&caps(), 60_000.0, &static_params());
2428 assert_eq!(e.phase(), TunePhase::Ramping);
2429 assert_eq!(p.upload_window, 1);
2430 }
2431
2432 #[test]
2433 fn upload_auto_level_benchmarks_once() {
2434 let mut e = TuningEngine::new(true, CompressLevel::Auto);
2435 e.begin_transfer(&caps(), 0.0, &static_params());
2436 let sample = vec![0xABu8; 256 * 1024];
2437 let lvl1 = e.upload_compress_level(&caps(), Some(&sample), 50.0);
2438 let lvl2 = e.upload_compress_level(&caps(), Some(&sample), 50.0);
2439 assert_eq!(lvl1, lvl2, "benchmark runs once, result cached");
2440 assert!(e.auto_level_ready());
2441 assert!((-8..=4).contains(&lvl1), "level inside advertised range, got {lvl1}");
2442 let mut e2 = TuningEngine::new(true, CompressLevel::Auto);
2444 e2.begin_transfer(&caps(), 0.0, &static_params());
2445 let lvl = e2.upload_compress_level(&caps(), Some(&[0u8; 0]), 50.0);
2446 assert_eq!(lvl, caps().compression.zrip_levels.default, "no sample → server default");
2447 }
2448
2449 #[test]
2450 fn disabled_engine_uses_static_params() {
2451 let mut e = TuningEngine::new(false, CompressLevel::Balanced);
2452 e.set_direction(TransferKind::Upload);
2453 let p = e.begin_transfer(&caps(), 0.0, &static_params());
2454 assert_eq!(p.concurrency, 4, "static config preserved");
2455 assert_eq!(p.upload_window, 8);
2456 assert_eq!(e.phase(), TunePhase::Uninitialized);
2457 assert!(e.tick(1_000.0, 12_500_000, Some(40.0), false).is_none());
2459 assert_eq!(p, static_params().clamped_into(&caps()));
2460 }
2461}