fix(http): disable pooling on shared reqwest client (#1092) (#1248)

## Overview

Fixes the remaining cases of
[#1092](#1092)
— customers continuing to see `Max retries exceeded, returning request
error` and `TRACES | Request failed: No requests sent` after the v94
patch in #1094.

PR #1094 disabled connection pooling on the **hyper** `HttpClient` used
by the trace flusher and stats flusher. It did not touch the **reqwest**
`Client` in `bottlecap/src/http.rs`, which is used by:

- the trace proxy (`proxy_flusher`) — the path AppSec and many tracer
configurations route through
- the metrics flusher
- the logs flusher

That client was still configured with `pool_idle_timeout(270s)` and
`tcp_keepalive(120s)`, leaving idle connections in the pool across
Lambda freeze/resume cycles. Connections go stale during a freeze (the
OS tears down the remote side while the local kernel still considers
them established), and the next request after resume fails with broken
pipe / RST → retries exhaust → the user-visible errors.

## Change

`bottlecap/src/http.rs`: replace `pool_idle_timeout(270s) +
tcp_keepalive(120s)` with `pool_max_idle_per_host(0)`. Mirrors the fix
in `bottlecap/src/traces/http_client.rs` and the pattern in libdatadog's
`new_client_periodic`. Each request opens a fresh connection; TLS
session resumption inside the client still amortizes the handshake.

Trade-off: one extra TLS handshake per flush. Matches the pre-v93
behavior that customers had no complaints about.

## Testing

- New unit test `shared_client_does_not_pool_connections` in
`bottlecap/src/http.rs`:
- Spawns a minimal HTTP/1.1 keep-alive server that counts accepted TCP
connections.
  - Issues two sequential GETs through `get_client(...)`.
  - Asserts the server saw **2** connections.
- Verified the test **fails** against the unpatched code (`opened=1`,
connection reused from pool) — proves it catches the regression.
- Verified the test **passes** against the patched code (`opened=2`,
fresh connection each time).
- `cargo test --lib` — 547 tests pass.
- `cargo clippy --lib --tests -- -D warnings` — clean.
- `cargo fmt --check` — clean.

## Other clients (unchanged)

| Client | File | Pooling | Notes |
|---|---|---|---|
| hyper `HttpClient` | `traces/http_client.rs` | already disabled
(#1094) | trace + stats flushers |
| reqwest `Client` (shared) | `bottlecap/src/http.rs` | **disabled
here** | trace proxy, metrics, logs |
| reqwest `Client` (Lambda Extensions API) | `bin/bottlecap/main.rs` |
unchanged | localhost-only; pooling irrelevant |

Author: duncanista

bottlecap/src/http.rs

@@ -26,9 +26,12 @@ pub fn get_client(config: &Arc<config::Config>) -> reqwest::Client {
 fn build_client(config: &Arc<config::Config>) -> Result<reqwest::Client, Box<dyn Error>> {
     let mut client = create_reqwest_client_builder()?
         .timeout(Duration::from_secs(config.flush_timeout))
-        .pool_idle_timeout(Some(Duration::from_secs(270)))
-        // Enable TCP keepalive
-        .tcp_keepalive(Some(Duration::from_secs(120)))
+        // Disable connection pooling to avoid stale connections after Lambda freeze/resume.
+        // The execution environment can be frozen for seconds to minutes between invocations;
+        // pooled connections become stale during this time and fail on reuse, surfacing as
+        // "Max retries exceeded" in the trace proxy / metrics / logs flushers. Mirrors the
+        // pattern in `traces/http_client.rs` and libdatadog's `new_client_periodic`.
+        .pool_max_idle_per_host(0)
         .danger_accept_invalid_certs(config.skip_ssl_validation);
 
     // Determine if we should use HTTP/1 or HTTP/2
@@ -145,3 +148,101 @@ pub fn headers_to_map(headers: &HeaderMap) -> HashMap<String, String> {
         })
         .collect()
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::TcpListener;
+
+    /// Spawns a minimal HTTP/1.1 keep-alive server that:
+    /// - increments `connection_counter` on every accepted TCP connection
+    /// - serves any number of requests on the same connection until the client
+    ///   closes it (matching how Datadog intakes behave in production).
+    ///
+    /// With pooling enabled, a client making two sequential requests will reuse
+    /// the same TCP connection → counter stays at 1. With pooling disabled,
+    /// each request opens a fresh connection → counter reaches 2.
+    async fn spawn_keepalive_server(connection_counter: Arc<AtomicUsize>) -> u16 {
+        let listener = TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("bind test listener");
+        let port = listener.local_addr().expect("listener local_addr").port();
+        tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener.accept().await else {
+                    return;
+                };
+                connection_counter.fetch_add(1, Ordering::SeqCst);
+                tokio::spawn(async move {
+                    let mut buf = vec![0u8; 8192];
+                    let mut accumulated = Vec::new();
+                    loop {
+                        let n = match sock.read(&mut buf).await {
+                            Ok(0) | Err(_) => return,
+                            Ok(n) => n,
+                        };
+                        accumulated.extend_from_slice(&buf[..n]);
+                        while let Some(idx) = accumulated.windows(4).position(|w| w == b"\r\n\r\n")
+                        {
+                            accumulated.drain(..idx + 4);
+                            if sock
+                                .write_all(
+                                    b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\nConnection: keep-alive\r\n\r\nok",
+                                )
+                                .await
+                                .is_err()
+                            {
+                                return;
+                            }
+                        }
+                    }
+                });
+            }
+        });
+        port
+    }
+
+    /// Verifies that the shared reqwest client does not pool idle connections:
+    /// two sequential requests to the same host must result in two distinct
+    /// TCP connections at the server. This guards against regression of
+    /// [issue #1092](https://github.com/DataDog/datadog-lambda-extension/issues/1092):
+    /// pooled connections go stale across Lambda freeze/resume and surface as
+    /// "Max retries exceeded" errors on the proxy / metrics / logs flushers.
+    #[tokio::test]
+    async fn shared_client_does_not_pool_connections() {
+        let counter = Arc::new(AtomicUsize::new(0));
+        let port = spawn_keepalive_server(counter.clone()).await;
+
+        let cfg = config::Config {
+            // Force HTTP/1 path; HTTP/2 would also work but H1 makes the pool
+            // behavior easiest to reason about with a hand-rolled server.
+            http_protocol: Some("http1".to_string()),
+            flush_timeout: 5,
+            ..config::Config::default()
+        };
+        let client = get_client(&Arc::new(cfg));
+
+        let url = format!("http://127.0.0.1:{port}/");
+
+        // First request: opens connection #1.
+        let r1 = client.get(&url).send().await.expect("first request");
+        assert_eq!(r1.status(), 200);
+        // Drain body so reqwest can decide whether to return the conn to the pool.
+        let _ = r1.bytes().await;
+
+        // Second request: with pooling disabled, must open a fresh connection #2.
+        // With pooling enabled (the v93-era regression), reqwest would reuse
+        // conn #1, which after a Lambda freeze would be stale.
+        let r2 = client.get(&url).send().await.expect("second request");
+        assert_eq!(r2.status(), 200);
+        let _ = r2.bytes().await;
+
+        let opened = counter.load(Ordering::SeqCst);
+        assert_eq!(
+            opened, 2,
+            "expected 2 fresh TCP connections, got {opened} (pooling not disabled?)"
+        );
+    }
+}