Updates; clippy fixes.

Author: hoytak

cas_client/src/adaptive_concurrency/controller.rs

@@ -239,7 +239,7 @@ impl ConnectionPermit {
     }
 
     pub(crate) async fn report_retryable_failure(&self) {
-        self.controller.clone().report_and_update(&self, None, false).await;
+        self.controller.clone().report_and_update(self, None, false).await;
     }
 }
 
@@ -248,8 +248,8 @@ impl ConnectionPermit {
 mod test_constants {
 
     pub const TR_HALF_LIFE_MS: u64 = 10;
-    pub const INCR_SPACING_MS: u64 = 4;
-    pub const DECR_SPACING_MS: u64 = 2;
+    pub const INCR_SPACING_MS: u64 = 200;
+    pub const DECR_SPACING_MS: u64 = 100;
 
     pub const TARGET_TIME_MS_S: u64 = 5;
     pub const TARGET_TIME_MS_L: u64 = 20;
@@ -303,10 +303,12 @@ mod tests {
 
         let controller = AdaptiveConcurrencyController::new_testing(1, (1, 4));
 
-        for _ in 0..10 {
+        for i in 0..10 {
             let permit = controller.acquire_connection_permit().await.unwrap();
-            advance(Duration::from_millis(1)).await;
+            // Increase the duration, so we're always going faster than predicted
+            advance(Duration::from_millis(12 - i)).await;
             permit.report_completion(B, true).await;
+
             advance(Duration::from_millis(INCR_SPACING_MS + 1)).await;
         }
 
@@ -324,19 +326,38 @@ mod tests {
         // Advance on so that the first success will trigger an adjustment.
         advance(Duration::from_millis(INCR_SPACING_MS + 1)).await;
 
-        let t = Instant::now();
+        for i in 0..5 {
+            let permit = controller.acquire_connection_permit().await.unwrap();
+            // Increase the duration, so we're always going faster than predicted
+            advance(Duration::from_millis(12 - i)).await;
+            permit.report_completion(B, true).await;
+
+            // Don't advance, so it should have only incremented by one as not enough time
+            // will have passed for more.
+        }
+
+        assert_eq!(controller.available_permits(), 2);
+        assert_eq!(controller.total_permits(), 2);
+
+        // Now, advance the clock by enough time to allow another change.
+        advance(Duration::from_millis(INCR_SPACING_MS + 1)).await;
 
-        while t.elapsed() < Duration::from_millis(INCR_SPACING_MS + 2) {
+        for i in 5..10 {
             let permit = controller.acquire_connection_permit().await.unwrap();
-            advance(Duration::from_millis(1)).await;
+            // Increase the duration, so we're always going faster than predicted
+            advance(Duration::from_millis(12 - i)).await;
             permit.report_completion(B, true).await;
+
+            // Don't advance, so it should have only incremented by one as not enough time
+            // will have passed for more.
         }
 
         // The window above should have had exactly two increases; one at the first success and one within the next
         // interval.
         assert_eq!(controller.available_permits(), 3);
         assert_eq!(controller.total_permits(), 3);
     }
+
     #[tokio::test]
     async fn test_permit_increase_on_slow_but_good_enough() {
         time::pause();

cas_client/src/adaptive_concurrency/latency_prediction.rs

@@ -78,14 +78,7 @@ impl LatencyPredictor {
     ///
     /// - `size_bytes`: the size of the completed transmission.
     /// - `duration`: the time taken to complete the transmission.
-    /// Updates the latency model with a new observation.
-    ///
-    /// Applies exponential decay to prior statistics and incorporates the new sample
-    /// using a numerically stable linear regression formula.
-    ///
-    /// - `size_bytes`: the size of the completed transmission.
-    /// - `duration`: the time taken to complete the transmission.
-    /// - `n_concurrent`: the number of concurrent connections at the time.
+    /// - `avg_concurrent`: an estimate of the average number of concurrent connections during transfer.
     pub fn update(&mut self, size_bytes: u64, duration: Duration, avg_concurrent: f64) {
         let now = Instant::now();
         let elapsed = now.duration_since(self.last_update).as_secs_f64();
@@ -188,9 +181,7 @@ impl LatencyPredictor {
         let query_bytes = 10 * 1024 * 1024;
 
         // How long would it take to transmit this at full bandwidth
-        let Some(min_latency) = self.predicted_latency(query_bytes, 1.) else {
-            return None;
-        };
+        let min_latency = self.predicted_latency(query_bytes, 1.)?;
 
         // Report bytes per sec in this model.
         Some(query_bytes as f64 / min_latency.max(1e-6))