fix: reject degenerate RTT predictions in adaptive concurrency

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<f64>, Option<f64>) {
-        // 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.

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