Fix panic when merging a stalled stream log to an empty LogBuffer

CHANGELOG.next.toml

@@ -9,4 +9,16 @@
 # message = "Fix typos in module documentation for generated crates"
 # references = ["smithy-rs#920"]
 # meta = { "breaking" = false, "tada" = false, "bug" = false, "target" = "client | server | all"}
-# author = "rcoh"
+# author = "rcoh"
+
+ [[aws-sdk-rust]]
+ message = "Fix bug where stalled stream protection would panic with an underflow if the first event was logged too soon."
+ references = ["smithy-rs#3744"]
+ meta = { "breaking" = false, "tada" = false, "bug" = true }
+ author = "Velfi"
+
+ [[smithy-rs]]
+ message = "Fix bug where stalled stream protection would panic with an underflow if the first event was logged too soon."
+ references = ["smithy-rs#3744"]
+ meta = { "breaking" = false, "tada" = false, "bug" = true, "target" = "client"}
+ author = "Velfi"

rust-runtime/aws-smithy-runtime/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "aws-smithy-runtime"
-version = "1.6.1"
+version = "1.6.2"
 authors = ["AWS Rust SDK Team <[email protected]>", "Zelda Hessler <[email protected]>"]
 description = "The new smithy runtime crate"
 edition = "2021"

rust-runtime/aws-smithy-runtime/src/client/http/body/minimum_throughput/throughput.rs

@@ -181,6 +181,7 @@ struct LogBuffer<const N: usize> {
     // polling gap. Once the length reaches N, it will never change again.
     length: usize,
 }
+
 impl<const N: usize> LogBuffer<N> {
     fn new() -> Self {
         Self {
@@ -247,6 +248,11 @@ impl<const N: usize> LogBuffer<N> {
         }
         counts
     }
+
+    /// If this LogBuffer is empty, returns `true`. Else, returns `false`.
+    fn is_empty(&self) -> bool {
+        return self.length == 0;
+    }
 }
 
 /// Report/summary of all the events in a time window.
@@ -334,7 +340,11 @@ impl ThroughputLogs {
 
     fn push(&mut self, now: SystemTime, value: Bin) {
         self.catch_up(now);
-        self.buffer.tail_mut().merge(value);
+        if self.buffer.is_empty() {
+            self.buffer.push(value)
+        } else {
+            self.buffer.tail_mut().merge(value);
+        }
         self.buffer.fill_gaps();
     }
 
@@ -552,4 +562,13 @@ mod test {
         let report = logs.report(start + Duration::from_millis(999));
         assert_eq!(ThroughputReport::Pending, report);
     }
+
+    #[test]
+    fn test_first_push_succeeds_although_time_window_has_not_elapsed() {
+        let t0 = SystemTime::UNIX_EPOCH;
+        let t1 = t0 + Duration::from_secs(1);
+        let mut tl = ThroughputLogs::new(Duration::from_secs(1), t1);
+
+        tl.push_pending(t0);
+    }
 }