fix(media-proxy): retry dropped connections and log the cause (#2749)

Cargo.lock

@@ -1815,6 +1815,7 @@ dependencies = [
  "http 1.4.2",
  "http-body 1.0.1",
  "http-body-util",
+ "hyper",
  "libc",
  "moka",
  "parking_lot",

fluxer_media_proxy/Cargo.toml

@@ -31,6 +31,7 @@ hmac = "0.13.0"
 http = "1.4.2"
 http-body = "1.0.1"
 http-body-util = "0.1.3"
+hyper = {version = "1.10.1", default-features = false}
 libc = "0.2.186"
 moka = {version = "0.12.15", features = ["sync"]}
 parking_lot = "0.12.5"

fluxer_media_proxy/src/http_client/retry.rs

@@ -6,6 +6,7 @@ use reqwest_middleware::Error as MiddlewareError;
 use reqwest_retry::{
     RetryDecision, RetryPolicy, Retryable, RetryableStrategy, policies::ExponentialBackoff,
 };
+use std::io::ErrorKind;
 use std::sync::Arc;
 use std::time::SystemTime;
 
@@ -87,7 +88,7 @@ fn retryable_error(error: &MiddlewareError) -> Option<Retryable> {
             #[cfg(target_arch = "wasm32")]
             let is_connect = false;
 
-            if error.is_timeout() || is_connect {
+            if error.is_timeout() || is_connect || is_dropped_connection(error) {
                 Some(Retryable::Transient)
             } else if error.is_body()
                 || error.is_decode()
@@ -103,11 +104,37 @@ fn retryable_error(error: &MiddlewareError) -> Option<Retryable> {
     }
 }
 
+fn is_dropped_connection(error: &(dyn std::error::Error + 'static)) -> bool {
+    let mut current = Some(error);
+    while let Some(error) = current {
+        if error
+            .downcast_ref::<hyper::Error>()
+            .is_some_and(hyper::Error::is_incomplete_message)
+        {
+            return true;
+        }
+        if error.downcast_ref::<std::io::Error>().is_some_and(|error| {
+            matches!(
+                error.kind(),
+                ErrorKind::ConnectionReset
+                    | ErrorKind::ConnectionAborted
+                    | ErrorKind::BrokenPipe
+                    | ErrorKind::UnexpectedEof
+            )
+        }) {
+            return true;
+        }
+        current = error.source();
+    }
+    false
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
     use crate::metrics::Metrics;
     use std::time::Duration;
+    use tokio::io::AsyncReadExt as _;
 
     fn response_with_status(status: StatusCode) -> Result<reqwest::Response, MiddlewareError> {
         let response = http::Response::builder()
@@ -256,4 +283,50 @@ mod tests {
             "fluxer_media_proxy_http_retries_exhausted_total 1"
         );
     }
+
+    #[derive(Debug, thiserror::Error)]
+    #[error("request failed")]
+    struct WrappedTransportError(#[source] std::io::Error);
+
+    #[test]
+    fn a_dropped_connection_is_recognised_anywhere_in_the_error_chain() {
+        for kind in [
+            ErrorKind::ConnectionReset,
+            ErrorKind::ConnectionAborted,
+            ErrorKind::BrokenPipe,
+            ErrorKind::UnexpectedEof,
+        ] {
+            assert!(
+                is_dropped_connection(&WrappedTransportError(std::io::Error::from(kind))),
+                "{kind:?} must count as a dropped connection"
+            );
+        }
+        assert!(!is_dropped_connection(&WrappedTransportError(
+            std::io::Error::from(ErrorKind::PermissionDenied)
+        )));
+    }
+
+    #[tokio::test]
+    async fn a_connection_closed_before_the_response_arrives_is_transient() {
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+            .await
+            .expect("a loopback listener binds");
+        let address = listener
+            .local_addr()
+            .expect("the listener has a local address");
+        tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.expect("the client connects");
+            let mut request = [0u8; 4096];
+            let _ = socket.read(&mut request).await;
+        });
+        let error = crate::http_client::build_raw_default()
+            .get(format!("http://{address}/"))
+            .send()
+            .await
+            .expect_err("the server closes the connection without answering");
+        assert!(matches!(
+            retryable_error(&MiddlewareError::Reqwest(error)),
+            Some(Retryable::Transient)
+        ));
+    }
 }

fluxer_media_proxy/src/storage/mod.rs

@@ -159,9 +159,9 @@ pub enum StorageError {
     S3(String),
     #[error(transparent)]
     Io(#[from] std::io::Error),
-    #[error(transparent)]
+    #[error("{}", error_with_causes(.0))]
     Http(#[from] reqwest::Error),
-    #[error(transparent)]
+    #[error("{}", error_with_causes(.0))]
     HttpMiddleware(#[from] reqwest_middleware::Error),
     #[error(transparent)]
     Sign(#[from] aws_sigv4::Error),
@@ -274,6 +274,17 @@ fn record_storage_outcome<T>(metrics: &StorageMetrics, result: &Result<T, Storag
     }
 }
 
+fn error_with_causes(error: &(dyn std::error::Error + 'static)) -> String {
+    let mut message = error.to_string();
+    let mut cause = error.source();
+    while let Some(error) = cause {
+        message.push_str(": ");
+        message.push_str(&error.to_string());
+        cause = error.source();
+    }
+    message
+}
+
 fn map_not_found(err: std::io::Error) -> StorageError {
     if err.kind() == std::io::ErrorKind::NotFound {
         StorageError::NotFound

fluxer_media_proxy/src/storage/tests/s3.rs

@@ -8,7 +8,14 @@ use crate::{
     storage::{ObjectReadRequest, ObjectStreamRequest, StorageError},
 };
 use http::{Method, StatusCode, header};
-use std::time::Duration;
+use std::{
+    sync::{
+        Arc,
+        atomic::{AtomicUsize, Ordering},
+    },
+    time::Duration,
+};
+use tokio::io::AsyncReadExt as _;
 
 const LAST_MODIFIED: &str = "Wed, 21 Oct 2015 07:28:00 GMT";
 
@@ -344,6 +351,72 @@ async fn stream_s3_surfaces_an_upstream_416_instead_of_a_storage_error() {
     assert_eq!(Some(0), object.content_length);
 }
 
+#[tokio::test]
+async fn a_read_on_a_dropped_connection_is_retried_on_a_new_one() {
+    let fake = fake_s3().await;
+    fake.put_object("cdn/a/b.txt", stored_object());
+    let (endpoint, accepted) = connection_dropping_front(fake.endpoint(), 1).await;
+    let tmp = tempfile::tempdir().unwrap();
+    let mut cfg = fake.config(tmp.path());
+    cfg.storage.s3_endpoint = endpoint;
+    let store = store(cfg);
+
+    let object = store.read_object("cdn", "a/b.txt").await.unwrap();
+
+    assert_eq!(b"hello world", &object.data[..]);
+    assert_eq!(2, accepted.load(Ordering::SeqCst));
+    assert_eq!(1, fake_gets(&fake, "/cdn/a/b.txt"));
+}
+
+#[tokio::test]
+async fn a_read_that_keeps_losing_its_connection_reports_the_cause() {
+    let fake = fake_s3().await;
+    fake.put_object("cdn/a/b.txt", stored_object());
+    let (endpoint, accepted) = connection_dropping_front(fake.endpoint(), usize::MAX).await;
+    let tmp = tempfile::tempdir().unwrap();
+    let mut cfg = fake.config(tmp.path());
+    cfg.storage.s3_endpoint = endpoint;
+    let store = store(cfg);
+
+    let error = store
+        .head_object("cdn", "a/b.txt")
+        .await
+        .expect_err("every connection to the origin is dropped");
+
+    assert!(
+        error
+            .to_string()
+            .contains("connection closed before message completed"),
+        "{error}"
+    );
+    assert_eq!(3, accepted.load(Ordering::SeqCst));
+    assert!(fake.requests().is_empty());
+}
+
+async fn connection_dropping_front(origin: &str, dropped: usize) -> (String, Arc<AtomicUsize>) {
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let endpoint = format!("http://{}", listener.local_addr().unwrap());
+    let origin = origin.trim_start_matches("http://").to_owned();
+    let accepted = Arc::new(AtomicUsize::new(0));
+    let counter = Arc::clone(&accepted);
+    tokio::spawn(async move {
+        while let Ok((mut client, _)) = listener.accept().await {
+            let index = counter.fetch_add(1, Ordering::SeqCst);
+            let origin = origin.clone();
+            tokio::spawn(async move {
+                if index < dropped {
+                    let mut request = [0u8; 4096];
+                    let _ = client.read(&mut request).await;
+                    return;
+                }
+                let mut upstream = tokio::net::TcpStream::connect(origin).await.unwrap();
+                let _ = tokio::io::copy_bidirectional(&mut client, &mut upstream).await;
+            });
+        }
+    });
+    (endpoint, accepted)
+}
+
 fn fake_gets(fake: &FakeS3, path: &str) -> usize {
     fake.requests()
         .iter()