fluxer/fluxer_media_proxy/src/http_client/retry.rs

// SPDX-License-Identifier: AGPL-3.0-or-later

use crate::metrics::http_client::HTTPClientMetrics;
use reqwest::StatusCode;
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::{Duration, SystemTime};

pub(super) struct ObservableRetryPolicy {
    inner: ExponentialBackoff,
    retry_window: Option<Duration>,
    metrics: Arc<HTTPClientMetrics>,
}

impl ObservableRetryPolicy {
    pub(super) fn new(
        inner: ExponentialBackoff,
        retry_window: Option<Duration>,
        metrics: Arc<HTTPClientMetrics>,
    ) -> Self {
        Self {
            inner,
            retry_window,
            metrics,
        }
    }
}

impl RetryPolicy for ObservableRetryPolicy {
    fn should_retry(&self, request_start_time: SystemTime, n_past_retries: u32) -> RetryDecision {
        let window_spent = self.retry_window.is_some_and(|window| {
            request_start_time
                .elapsed()
                .is_ok_and(|elapsed| elapsed >= window)
        });
        let decision = if window_spent {
            RetryDecision::DoNotRetry
        } else {
            self.inner.should_retry(request_start_time, n_past_retries)
        };
        match decision {
            RetryDecision::Retry { .. } => self.metrics.record_retry(),
            RetryDecision::DoNotRetry => self.metrics.record_retries_exhausted(),
        }
        decision
    }
}

pub(super) struct MediaProxyRetryStrategy {
    metrics: Arc<HTTPClientMetrics>,
}

impl MediaProxyRetryStrategy {
    pub(super) fn new(metrics: Arc<HTTPClientMetrics>) -> Self {
        Self { metrics }
    }
}

impl RetryableStrategy for MediaProxyRetryStrategy {
    fn handle(&self, result: &Result<reqwest::Response, MiddlewareError>) -> Option<Retryable> {
        match result {
            Ok(response) => retryable_status(response.status()).map(|retryable| {
                if retryable == Retryable::Transient {
                    self.metrics.record_retryable_status();
                }
                retryable
            }),
            Err(error) => retryable_error(error).map(|retryable| {
                if retryable == Retryable::Transient {
                    self.metrics.record_retryable_error();
                }
                retryable
            }),
        }
    }
}

fn retryable_status(status: StatusCode) -> Option<Retryable> {
    match status {
        StatusCode::REQUEST_TIMEOUT
        | StatusCode::TOO_MANY_REQUESTS
        | StatusCode::INTERNAL_SERVER_ERROR
        | StatusCode::BAD_GATEWAY
        | StatusCode::SERVICE_UNAVAILABLE
        | StatusCode::GATEWAY_TIMEOUT => Some(Retryable::Transient),
        status if status.is_client_error() || status.is_server_error() => Some(Retryable::Fatal),
        _ => None,
    }
}

fn retryable_error(error: &MiddlewareError) -> Option<Retryable> {
    // A pinned-resolver rejection is a policy decision, not a transient fault, and retrying it only
    // repeats the same lookup against a name the caller supplied.
    if crate::public_net_policy::is_pinned_dns_failure(error) {
        return Some(Retryable::Fatal);
    }
    match error {
        MiddlewareError::Middleware(_) => Some(Retryable::Fatal),
        MiddlewareError::Reqwest(error) => {
            #[cfg(not(target_arch = "wasm32"))]
            let is_connect = error.is_connect();
            #[cfg(target_arch = "wasm32")]
            let is_connect = false;

            if error.is_timeout() || is_connect || is_dropped_connection(error) {
                Some(Retryable::Transient)
            } else if error.is_body()
                || error.is_decode()
                || error.is_builder()
                || error.is_redirect()
                || error.is_status()
            {
                Some(Retryable::Fatal)
            } else {
                None
            }
        }
    }
}

pub(crate) 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()
            .status(status)
            .body("")
            .expect("building a synthetic response cannot fail");
        Ok(reqwest::Response::from(response))
    }

    fn counter_line(rendered: &str, name: &str) -> String {
        rendered
            .lines()
            .find(|line| line.starts_with(name) && !line.starts_with('#'))
            .unwrap_or_else(|| panic!("{name} is missing from the rendered metrics"))
            .to_owned()
    }

    #[test]
    fn retry_strategy_retries_only_explicit_transient_statuses() {
        assert!(matches!(
            retryable_status(StatusCode::REQUEST_TIMEOUT),
            Some(Retryable::Transient)
        ));
        assert!(matches!(
            retryable_status(StatusCode::TOO_MANY_REQUESTS),
            Some(Retryable::Transient)
        ));
        assert!(matches!(
            retryable_status(StatusCode::INTERNAL_SERVER_ERROR),
            Some(Retryable::Transient)
        ));
        assert!(matches!(
            retryable_status(StatusCode::BAD_GATEWAY),
            Some(Retryable::Transient)
        ));
        assert!(matches!(
            retryable_status(StatusCode::SERVICE_UNAVAILABLE),
            Some(Retryable::Transient)
        ));
        assert!(matches!(
            retryable_status(StatusCode::GATEWAY_TIMEOUT),
            Some(Retryable::Transient)
        ));
        assert!(matches!(
            retryable_status(StatusCode::NOT_FOUND),
            Some(Retryable::Fatal)
        ));
        assert!(matches!(
            retryable_status(StatusCode::NOT_IMPLEMENTED),
            Some(Retryable::Fatal)
        ));
        assert!(retryable_status(StatusCode::OK).is_none());
        assert!(retryable_status(StatusCode::FOUND).is_none());
    }

    #[test]
    fn retryable_status_ignores_success_and_redirect_statuses() {
        assert!(retryable_status(StatusCode::OK).is_none());
        assert!(retryable_status(StatusCode::FOUND).is_none());
    }

    #[test]
    fn server_errors_outside_the_whitelist_stay_fatal() {
        for status in [
            StatusCode::NOT_IMPLEMENTED,
            StatusCode::HTTP_VERSION_NOT_SUPPORTED,
            StatusCode::INSUFFICIENT_STORAGE,
            StatusCode::LOOP_DETECTED,
            StatusCode::NETWORK_AUTHENTICATION_REQUIRED,
        ] {
            assert!(
                matches!(retryable_status(status), Some(Retryable::Fatal)),
                "{status} must not be retried"
            );
        }
    }

    #[test]
    fn middleware_failures_are_fatal_and_are_not_counted_as_transient() {
        let metrics = Metrics::new();
        let strategy = MediaProxyRetryStrategy::new(metrics.http_client());
        let error = Err(MiddlewareError::Middleware(anyhow::anyhow!(
            "middleware refused the request"
        )));
        assert!(matches!(strategy.handle(&error), Some(Retryable::Fatal)));
        let rendered = metrics.render();
        assert_eq!(
            counter_line(
                &rendered,
                "fluxer_media_proxy_http_retryable_classifications_total{reason=\"error\"}"
            ),
            "fluxer_media_proxy_http_retryable_classifications_total{reason=\"error\"} 0"
        );
    }

    #[test]
    fn transient_status_classifications_are_counted_once_and_fatal_ones_are_not() {
        let metrics = Metrics::new();
        let strategy = MediaProxyRetryStrategy::new(metrics.http_client());
        assert!(matches!(
            strategy.handle(&response_with_status(StatusCode::SERVICE_UNAVAILABLE)),
            Some(Retryable::Transient)
        ));
        assert!(matches!(
            strategy.handle(&response_with_status(StatusCode::NOT_FOUND)),
            Some(Retryable::Fatal)
        ));
        assert!(
            strategy
                .handle(&response_with_status(StatusCode::OK))
                .is_none()
        );
        let rendered = metrics.render();
        assert_eq!(
            counter_line(
                &rendered,
                "fluxer_media_proxy_http_retryable_classifications_total{reason=\"status\"}"
            ),
            "fluxer_media_proxy_http_retryable_classifications_total{reason=\"status\"} 1"
        );
    }

    #[test]
    fn every_retry_decision_reaches_the_injected_metrics_handle() {
        let metrics = Metrics::new();
        let backoff = ExponentialBackoff::builder()
            .retry_bounds(Duration::from_millis(1), Duration::from_millis(2))
            .build_with_max_retries(1);
        let policy = ObservableRetryPolicy::new(backoff, None, metrics.http_client());
        let started = SystemTime::now();
        assert!(matches!(
            policy.should_retry(started, 0),
            RetryDecision::Retry { .. }
        ));
        assert!(matches!(
            policy.should_retry(started, 1),
            RetryDecision::DoNotRetry
        ));
        let rendered = metrics.render();
        assert_eq!(
            counter_line(&rendered, "fluxer_media_proxy_http_retries_total"),
            "fluxer_media_proxy_http_retries_total 1"
        );
        assert_eq!(
            counter_line(&rendered, "fluxer_media_proxy_http_retries_exhausted_total"),
            "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)
        ));
    }

    fn policy(window: Option<Duration>) -> ObservableRetryPolicy {
        ObservableRetryPolicy::new(
            ExponentialBackoff::builder().build_with_max_retries(2),
            window,
            Arc::new(HTTPClientMetrics::new()),
        )
    }

    #[test]
    fn a_request_past_its_retry_window_is_not_retried() {
        let started = SystemTime::now() - Duration::from_secs(20);
        assert!(matches!(
            policy(Some(Duration::from_secs(15))).should_retry(started, 0),
            RetryDecision::DoNotRetry
        ));
    }

    #[test]
    fn a_request_inside_its_retry_window_is_retried() {
        assert!(matches!(
            policy(Some(Duration::from_secs(15))).should_retry(SystemTime::now(), 0),
            RetryDecision::Retry { .. }
        ));
    }

    #[test]
    fn a_policy_without_a_window_keeps_retrying_slow_requests() {
        let started = SystemTime::now() - Duration::from_secs(60);
        assert!(matches!(
            policy(None).should_retry(started, 0),
            RetryDecision::Retry { .. }
        ));
    }
}