Fix panic when merging a stalled stream log to an empty LogBuffer (#3744)

## Motivation and Context
<!--- Why is this change required? What problem does it solve? -->
<!--- If it fixes an open issue, please link to the issue here -->
this fixes an issue submitted by an internal user.

## Description
<!--- Describe your changes in detail -->
If the first event pushed into the log underlying stalled stream
protection, then the log manager would panic. This is because it was
expecting to merge the new event in but there was nothing to merge it
with.

To fix this, I updated the code to `push` the event when the log manager
is empty, instead of merging it.

## Testing
<!--- Please describe in detail how you tested your changes -->
<!--- Include details of your testing environment, and the tests you ran
to -->
<!--- see how your change affects other areas of the code, etc. -->
ran existing tests and wrote a new test exercising the
previous-panicking code.

## Checklist
<!--- If a checkbox below is not applicable, then please DELETE it
rather than leaving it unchecked -->
- [x] I have updated `CHANGELOG.next.toml` if I made changes to the
smithy-rs codegen or runtime crates
- [x] I have updated `CHANGELOG.next.toml` if I made changes to the AWS
SDK, generated SDK code, or SDK runtime crates

----

_By submitting this pull request, I confirm that you can use, modify,
copy, and redistribute this contribution, under the terms of your
choice._

Author: Zelda Hessler

CHANGELOG.next.toml

@@ -14,5 +14,17 @@
 [[smithy-rs]]
 message = "Support `stringArray` type in endpoints params"
 references = ["smithy-rs#3742"]
-meta = { "breaking" = false, "tada" = false, "bug" = false, "target" = "client"}
+meta = { "breaking" = false, "tada" = false, "bug" = false, "target" = "client" }
 author = "landonxjames"
+
+[[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/Cargo.lock

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

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

@@ -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 {
+        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);
+    }
 }