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);