From 9dfcc24548b01e3473cd8d96ca71e112960241a9 Mon Sep 17 00:00:00 2001 From: Muhammad Maaz Date: Sat, 5 Sep 2026 13:11:52 +0100 Subject: [PATCH 1/2] fix: reject degenerate RTT predictions --- .../adaptive_concurrency/exp_weighted_olr.rs | 33 ++++++++++++++++--- .../adaptive_concurrency/rtt_prediction.rs | 28 +++++++++++++++- 2 files changed, 55 insertions(+), 6 deletions(-) diff --git a/xet_client/src/cas_client/adaptive_concurrency/exp_weighted_olr.rs b/xet_client/src/cas_client/adaptive_concurrency/exp_weighted_olr.rs index 946cfdc9a..6fe61ba3c 100644 --- a/xet_client/src/cas_client/adaptive_concurrency/exp_weighted_olr.rs +++ b/xet_client/src/cas_client/adaptive_concurrency/exp_weighted_olr.rs @@ -65,15 +65,17 @@ impl ExpWeightedOnlineLinearRegression { /// Predict the mean and standard deviation of the *fitted mean* at x. /// /// Returns: - /// - (None, None) if the model is not identifiable yet (normal matrix singular). + /// - (None, None) if the model is not identifiable yet (normal matrix singular or ill-conditioned). /// - (Some(mean), None) if coefficients can be estimated but df <= 0, so we can't compute a standard deviation /// yet. /// - (Some(mean), Some(std_dev)) otherwise. pub fn predict(&self, x0: f64) -> (Option, Option) { - // Need a well-conditioned normal matrix to estimate beta. + // Need a well-conditioned normal matrix to estimate beta. Use a relative + // threshold so near-collinear x values are rejected independent of scale. let delta = self.sw * self.sxx - self.sx * self.sx; - if delta.abs() < 1e-12 { - // Can't estimate beta0/beta1 at all. + let scale = (self.sw * self.sxx).abs(); + if scale == 0.0 || delta <= 1e-12 * scale { + // Can't estimate beta0/beta1 reliably. return (None, None); } @@ -114,7 +116,8 @@ impl ExpWeightedOnlineLinearRegression { #[allow(dead_code)] pub fn coefficients(&self) -> Option<(f64, f64)> { let delta = self.sw * self.sxx - self.sx * self.sx; - if delta.abs() < 1e-12 { + let scale = (self.sw * self.sxx).abs(); + if scale == 0.0 || delta <= 1e-12 * scale { return None; } @@ -321,6 +324,26 @@ mod tests { assert_abs_diff_eq!(mean2.unwrap(), mean_off.unwrap(), epsilon = 1e-10); } + #[test] + fn test_near_singular_x_rejected() { + let mut model = ExpWeightedOnlineLinearRegression::new(1000.0); + + // The determinant is non-zero in absolute terms, but x has effectively no + // spread relative to its magnitude. The old absolute threshold accepted this. + for (x, y) in [ + (64.0, 1.0000), + (64.00001, 1.0001), + (64.00002, 0.9999), + (64.00003, 1.0002), + ] { + model.update(1.0, x, y); + } + + let (mean, std_dev) = model.predict(10.0); + assert!(mean.is_none()); + assert!(std_dev.is_none()); + } + #[test] fn test_singular_x_all_same() { // All x are the same -> normal equations matrix is singular. diff --git a/xet_client/src/cas_client/adaptive_concurrency/rtt_prediction.rs b/xet_client/src/cas_client/adaptive_concurrency/rtt_prediction.rs index bd05a250f..d0ed7fcdd 100644 --- a/xet_client/src/cas_client/adaptive_concurrency/rtt_prediction.rs +++ b/xet_client/src/cas_client/adaptive_concurrency/rtt_prediction.rs @@ -117,8 +117,14 @@ impl RTTPredictor { // How long would it take to transmit this at full bandwidth let min_rtt = self.predicted_rtt(query_bytes, 1.)?; + // A non-positive RTT is not a physically meaningful bandwidth estimate. In + // particular, predicted_rtt clamps negative regression output to zero. + if min_rtt <= 0.0 { + return None; + } + // Report bytes per sec in this model. - Some(query_bytes as f64 / min_rtt.max(1e-6)) + Some(query_bytes as f64 / min_rtt) } /// Computes the quantile (0.0 to 1.0) of an observed RTT under the predicted normal distribution. @@ -281,6 +287,26 @@ mod tests { assert!(se >= 0.0); } + #[test] + fn test_predicted_bandwidth_rejects_non_positive_rtt() { + let mut predictor = RTTPredictor::new(1000.0); + + // Fit a decreasing relationship whose extrapolated RTT at 10 MiB is negative. + // predicted_rtt() clamps it to zero, so bandwidth must not turn that clamp + // into an enormous synthetic throughput estimate. + for (size_mb, duration_secs) in [(1, 3.0), (2, 2.0), (3, 1.0), (4, 0.1)] { + predictor.update( + size_mb * 1024 * 1024, + Duration::from_secs_f64(duration_secs), + 1.0, + 1.0, + ); + } + + assert_eq!(predictor.predicted_rtt(10 * 1024 * 1024, 1.0), Some(0.0)); + assert!(predictor.predicted_bandwidth().is_none()); + } + #[test] fn test_rtt_quantile() { let mut predictor = RTTPredictor::new(10.0); From c3a0d74fab69207306dada96c6e4792e5066665d Mon Sep 17 00:00:00 2001 From: Muhammad Maaz Date: Sun, 13 Sep 2026 20:09:18 +0100 Subject: [PATCH 2/2] test: cover near-singular RTT controller behaviour --- .../adaptive_concurrency/controller.rs | 33 +++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/xet_client/src/cas_client/adaptive_concurrency/controller.rs b/xet_client/src/cas_client/adaptive_concurrency/controller.rs index 906cdcc41..871af9469 100644 --- a/xet_client/src/cas_client/adaptive_concurrency/controller.rs +++ b/xet_client/src/cas_client/adaptive_concurrency/controller.rs @@ -869,6 +869,39 @@ mod tests { } } + #[tokio::test] + async fn test_near_singular_rtt_fit_does_not_increase_concurrency() { + time::pause(); + + let controller = AdaptiveConcurrencyController::new_testing(2, (1, 4)); + let base_size = 32 * 1024 * 1024; + + { + let mut state = controller.state.lock().await; + + // Nearly identical transfer sizes with small RTT jitter produce an + // unstable negative slope. Extrapolating that fit to the reference + // size at the next concurrency level would clamp the RTT to zero. + for (offset, duration) in [(0, 1.3), (5, 1.2), (10, 1.1), (15, 1.0)] { + let size = base_size + offset; + state.rtt_predictor.update(size, Duration::from_secs_f64(duration), 2.0, 1.0); + state.update_size_tracking(size); + state.update_success(true, 1.0); + } + } + + advance(Duration::from_millis(INCR_SPACING_MS + 1)).await; + + let _first_permit = controller.acquire_connection_permit().await.unwrap(); + let permit = controller.acquire_connection_permit().await.unwrap(); + advance(Duration::from_secs(1)).await; + permit.report_completion(base_size + 20, true).await; + + // An unidentifiable fit must not be treated as a zero-RTT prediction and + // used to approve an increase from two permits to three. + assert_eq!(controller.total_permits(), 2); + } + #[tokio::test] async fn test_permit_decrease_on_explicit_failure() { time::pause();