fix(sidecar): preserve terminal metadata and add isolated unit tests

lib/sidecar/common/Cargo.toml

@@ -34,10 +34,11 @@ tonic-v14 = { workspace = true, optional = true }
 tracing = { workspace = true }
 url = { workspace = true }
 
-[features]
-# The v14 protobuf/tonic surface, for sidecars generated against prost 0.14.
-tonic-v14 = ["dep:tonic-v14", "dep:prost-types-v14"]
-
 [dev-dependencies]
 async-trait = { workspace = true }
 temp-env = { version = "0.3.6", features = ["async_closure"] }
+tokio = { workspace = true, features = ["test-util"] }
+
+[features]
+# The v14 protobuf/tonic surface, for sidecars generated against prost 0.14.
+tonic-v14 = ["dep:tonic-v14", "dep:prost-types-v14"]

lib/sidecar/common/src/error.rs

@@ -129,6 +129,21 @@ mod tests {
     fn maps_transport_statuses_to_backend_errors() {
         for (code, expected) in [
             (tonic::Code::InvalidArgument, BackendError::InvalidArgument),
+            (tonic::Code::NotFound, BackendError::InvalidArgument),
+            (tonic::Code::OutOfRange, BackendError::InvalidArgument),
+            (
+                tonic::Code::FailedPrecondition,
+                BackendError::InvalidArgument,
+            ),
+            (tonic::Code::AlreadyExists, BackendError::InvalidArgument),
+            (tonic::Code::Unknown, BackendError::Unknown),
+            (tonic::Code::Unimplemented, BackendError::Unknown),
+            (tonic::Code::ResourceExhausted, BackendError::Unknown),
+            (tonic::Code::PermissionDenied, BackendError::Unknown),
+            (tonic::Code::Unauthenticated, BackendError::Unknown),
+            (tonic::Code::Aborted, BackendError::Unknown),
+            (tonic::Code::DataLoss, BackendError::Unknown),
+            (tonic::Code::Ok, BackendError::Unknown),
             (tonic::Code::Unavailable, BackendError::CannotConnect),
             (tonic::Code::Cancelled, BackendError::Cancelled),
             (
@@ -145,6 +160,17 @@ mod tests {
         ] {
             let error = status_to_dynamo("Test", tonic::Status::new(code, "failure"));
             assert_eq!(error.error_type(), ErrorType::Backend(expected));
+            assert!(error.to_string().contains("Test: failure"));
+            assert!(error.to_string().contains(&format!("{code:?}")));
+            #[cfg(feature = "tonic-v14")]
+            {
+                let v14 = crate::error::status_to_dynamo_v14(
+                    "Test",
+                    tonic_v14::Status::new(tonic_v14::Code::from_i32(code as i32), "failure"),
+                );
+                assert_eq!(v14.error_type(), error.error_type());
+                assert_eq!(v14.to_string(), error.to_string());
+            }
         }
     }
 }

lib/sidecar/common/src/json.rs

@@ -173,6 +173,7 @@ pub use v14::{json_to_struct as json_to_struct_v14, struct_to_json as struct_to_
 
 #[cfg(test)]
 mod tests {
+    use prost_types::value::Kind;
     use serde_json::json;
 
     use super::{json_to_struct, struct_to_json};
@@ -221,4 +222,58 @@ mod tests {
         assert_eq!(v13, payload);
         assert_eq!(v14, payload);
     }
+
+    #[test]
+    fn integer_boundaries_and_negative_inexact_values_are_checked() {
+        for number in [-(1_i64 << 53), -1, 0, 1, 1_i64 << 53] {
+            let value = json!({"nested": [number, {"fraction": -3.25}]});
+            assert_eq!(
+                struct_to_json(
+                    json_to_struct(value.clone(), "payload").unwrap(),
+                    "test-engine",
+                    "payload"
+                )
+                .unwrap(),
+                value
+            );
+        }
+        for number in [i64::MIN, -(1_i64 << 53) - 1, (1_i64 << 53) + 1, i64::MAX] {
+            let error = json_to_struct(json!({"nested": [number]}), "payload").unwrap_err();
+            assert_eq!(
+                error.error_type(),
+                dynamo_backend_common::ErrorType::Backend(
+                    dynamo_backend_common::BackendError::InvalidArgument
+                )
+            );
+            assert!(error.to_string().contains("cannot be represented exactly"));
+        }
+    }
+
+    #[test]
+    fn incoming_nonfinite_numbers_fail_inside_nested_payloads() {
+        for value in [f64::NAN, f64::INFINITY, f64::NEG_INFINITY] {
+            let wire = prost_types::Struct {
+                fields: [(
+                    "nested".into(),
+                    prost_types::Value {
+                        kind: Some(Kind::ListValue(prost_types::ListValue {
+                            values: vec![prost_types::Value {
+                                kind: Some(Kind::NumberValue(value)),
+                            }],
+                        })),
+                    },
+                )]
+                .into_iter()
+                .collect(),
+            };
+            let error = struct_to_json(wire, "test-engine", "payload").unwrap_err();
+            assert_eq!(
+                error.error_type(),
+                dynamo_backend_common::ErrorType::Backend(
+                    dynamo_backend_common::BackendError::Unknown
+                )
+            );
+            assert!(error.to_string().contains("NaN or infinity"));
+        }
+    }
 }

lib/sidecar/common/src/lib.rs

@@ -34,3 +34,7 @@ pub use transport::{
 };
 
 pub use run::run;
+
+#[cfg(test)]
+#[path = "transport/tests.rs"]
+mod transport_tests;

lib/sidecar/common/src/transport.rs

@@ -35,40 +35,17 @@ impl GrpcChannelPool {
         let tonic_endpoint = Endpoint::from_shared(endpoint_label.clone()).map_err(|error| {
             invalid_argument(format!("invalid {peer} endpoint after validation: {error}"))
         })?;
-        let deadline = checked_instant_add(
-            Instant::now(),
-            transport.startup_deadline,
-            "gRPC startup deadline",
-        )?;
-        let first = connect_until_ready(
+        let channels = connect_pool_with(
             peer,
-            tonic_endpoint.clone(),
-            endpoint_label.clone(),
-            1,
+            &endpoint_label,
             transport,
-            deadline,
             bootstrap,
+            |_, attempt_timeout| {
+                let endpoint = tonic_endpoint.clone().connect_timeout(attempt_timeout);
+                async move { endpoint.connect().await }
+            },
         )
         .await?;
-        let mut channels = vec![first];
-        let remaining = try_join_all((1..transport.connections.get()).map(|index| {
-            let endpoint = tonic_endpoint.clone();
-            let endpoint_label = endpoint_label.clone();
-            async move {
-                connect_until_ready(
-                    peer,
-                    endpoint,
-                    endpoint_label,
-                    index + 1,
-                    transport,
-                    deadline,
-                    bootstrap,
-                )
-                .await
-            }
-        }))
-        .await?;
-        channels.extend(remaining);
         Ok(Self {
             channels,
             next: AtomicUsize::new(0),
@@ -89,15 +66,63 @@ impl GrpcChannelPool {
     }
 }
 
-async fn connect_until_ready(
+pub(crate) async fn connect_pool_with<C, F, E>(
+    peer: &str,
+    endpoint: &str,
+    transport: GrpcTransportConfig,
+    bootstrap: bool,
+    connect: impl Fn(usize, Duration) -> F,
+) -> Result<Vec<C>, DynamoError>
+where
+    F: std::future::Future<Output = Result<C, E>>,
+    E: std::error::Error + 'static,
+{
+    let deadline = checked_instant_add(
+        Instant::now(),
+        transport.startup_deadline,
+        "gRPC startup deadline",
+    )?;
+    let first = connect_until_ready(
+        peer,
+        endpoint,
+        1,
+        transport,
+        deadline,
+        bootstrap,
+        |timeout| connect(1, timeout),
+    )
+    .await?;
+    let mut channels = vec![first];
+    let remaining = try_join_all((2..=transport.connections.get()).map(|slot| {
+        let connect = &connect;
+        connect_until_ready(
+            peer,
+            endpoint,
+            slot,
+            transport,
+            deadline,
+            bootstrap,
+            move |timeout| connect(slot, timeout),
+        )
+    }))
+    .await?;
+    channels.extend(remaining);
+    Ok(channels)
+}
+
+async fn connect_until_ready<C, F, E>(
     peer: &str,
-    endpoint: Endpoint,
-    endpoint_label: String,
+    endpoint_label: &str,
     pool_slot: usize,
     transport: GrpcTransportConfig,
     deadline: Instant,
     bootstrap: bool,
-) -> Result<Channel, DynamoError> {
+    connect: impl Fn(Duration) -> F,
+) -> Result<C, DynamoError>
+where
+    F: std::future::Future<Output = Result<C, E>>,
+    E: std::error::Error + 'static,
+{
     let started = Instant::now();
     let mut attempt = 0_u64;
     let mut last_error = None;
@@ -109,7 +134,7 @@ async fn connect_until_ready(
         if remaining.is_zero() {
             return Err(startup_timeout(
                 peer,
-                &endpoint_label,
+                endpoint_label,
                 pool_slot,
                 attempt,
                 started.elapsed(),
@@ -119,10 +144,12 @@ async fn connect_until_ready(
         }
 
         attempt += 1;
-        let attempt_endpoint = endpoint
-            .clone()
-            .connect_timeout(transport.connect_attempt_timeout.min(remaining));
-        match timeout_at(deadline, attempt_endpoint.connect()).await {
+        match timeout_at(
+            deadline,
+            connect(transport.connect_attempt_timeout.min(remaining)),
+        )
+        .await
+        {
             Ok(Ok(channel)) => return Ok(channel),
             Ok(Err(error)) => {
                 let detailed_error = format_error_chain(&error);
@@ -165,7 +192,7 @@ async fn connect_until_ready(
             Err(_) => {
                 return Err(startup_timeout(
                     peer,
-                    &endpoint_label,
+                    endpoint_label,
                     pool_slot,
                     attempt,
                     started.elapsed(),
@@ -178,7 +205,7 @@ async fn connect_until_ready(
         if Instant::now() >= deadline {
             return Err(startup_timeout(
                 peer,
-                &endpoint_label,
+                endpoint_label,
                 pool_slot,
                 attempt,
                 started.elapsed(),

lib/sidecar/common/src/transport/tests.rs

@@ -0,0 +1,104 @@
+// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
+// SPDX-License-Identifier: Apache-2.0
+
+use std::cell::Cell;
+use std::io;
+use std::num::NonZeroUsize;
+use std::time::Duration;
+
+use crate::GrpcTransportConfig;
+use crate::transport::connect_pool_with;
+use dynamo_backend_common::{BackendError, ErrorType};
+use tokio::time::{Instant, sleep};
+
+fn config() -> GrpcTransportConfig {
+    GrpcTransportConfig {
+        connections: NonZeroUsize::new(3).unwrap(),
+        connect_attempt_timeout: Duration::from_millis(80),
+        retry_interval: Duration::from_millis(10),
+        startup_deadline: Duration::from_millis(100),
+    }
+}
+
+#[tokio::test(start_paused = true)]
+async fn failed_connection_retries_then_pool_contains_every_slot() {
+    let attempts = Cell::new(0);
+    let started = Instant::now();
+    let channels = connect_pool_with("test", "in-memory", config(), false, |slot, timeout| {
+        assert_eq!(timeout, Duration::from_millis(80));
+        let attempt = attempts.get();
+        attempts.set(attempt + 1);
+        async move {
+            if attempt == 0 {
+                Err(io::Error::other("initial failure"))
+            } else {
+                Ok(slot)
+            }
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(channels, vec![1, 2, 3]);
+    assert_eq!(attempts.get(), 4);
+    assert_eq!(started.elapsed(), Duration::from_millis(10));
+}
+
+#[tokio::test(start_paused = true)]
+async fn later_pool_slots_share_the_original_deadline() {
+    let observed = std::cell::RefCell::new(Vec::new());
+    let started = Instant::now();
+    let error = connect_pool_with("test", "in-memory", config(), false, |slot, timeout| {
+        observed.borrow_mut().push((slot, timeout));
+        async move {
+            if slot == 1 {
+                sleep(Duration::from_millis(70)).await;
+                Ok(slot)
+            } else {
+                std::future::pending::<Result<usize, io::Error>>().await
+            }
+        }
+    })
+    .await
+    .unwrap_err();
+    assert_eq!(started.elapsed(), Duration::from_millis(100));
+    assert_eq!(
+        *observed.borrow(),
+        vec![
+            (1, Duration::from_millis(80)),
+            (2, Duration::from_millis(30)),
+            (3, Duration::from_millis(30))
+        ]
+    );
+    assert_eq!(
+        error.error_type(),
+        ErrorType::Backend(BackendError::CannotConnect)
+    );
+    assert!(error.to_string().contains("pool slot 2"));
+    assert!(error.to_string().contains("exceeded the startup deadline"));
+}
+
+#[tokio::test(start_paused = true)]
+async fn retry_wait_is_capped_and_last_failure_is_preserved() {
+    let mut transport = config();
+    transport.retry_interval = Duration::from_millis(60);
+    let attempts = Cell::new(0);
+    let started = Instant::now();
+    let error = connect_pool_with("test", "in-memory", transport, false, |_, _| {
+        let attempt = attempts.get();
+        attempts.set(attempt + 1);
+        async move {
+            Err::<(), _>(io::Error::other(if attempt == 0 {
+                "initial connection failure"
+            } else {
+                "latest connection failure"
+            }))
+        }
+    })
+    .await
+    .unwrap_err();
+    assert_eq!(started.elapsed(), Duration::from_millis(100));
+    assert_eq!(attempts.get(), 2);
+    assert!(error.to_string().contains("latest connection failure"));
+    assert!(!error.to_string().contains("initial connection failure"));
+    assert!(error.to_string().contains("after 2 attempts"));
+}