saluki_io/net/util/retry/lifecycle/
http.rs

1use std::{borrow::Cow, fmt, io, time::Duration};
2
3use http::StatusCode;
4use tower::timeout::error::Elapsed;
5use tracing::{debug, warn};
6
7use super::{RetryCauseTelemetry, RetryLifecycle};
8
9/// A standard HTTP retry lifecycle that emits contextual information about HTTP requests and responses..
10///
11/// This lifecycle emits user-friendly logs about retry attempts, including the request URI and response code. It
12/// provides additional destructuring/introspection of errors to surface contextual information such as requests failing
13/// due to DNS, connection errors, TLS, and so on.
14///
15/// The same categorization drives telemetry when [`RetryCauseTelemetry`] is attached with
16/// [`with_telemetry`][Self::with_telemetry], so the logs and the metrics can't disagree about why a request was
17/// retried.
18#[derive(Clone, Default)]
19pub struct StandardHttpRetryLifecycle {
20    telemetry: Option<RetryCauseTelemetry>,
21}
22
23impl StandardHttpRetryLifecycle {
24    /// Creates a new `StandardHttpRetryLifecycle` that only logs retries.
25    pub fn new() -> Self {
26        Self::default()
27    }
28
29    /// Counts retries by cause, in addition to logging them.
30    pub fn with_telemetry(mut self, telemetry: RetryCauseTelemetry) -> Self {
31        self.telemetry = Some(telemetry);
32        self
33    }
34}
35
36impl<B, B2, E> RetryLifecycle<http::Request<B>, http::Response<B2>, E> for StandardHttpRetryLifecycle
37where
38    E: DynError,
39{
40    fn before_retry(
41        &self, req: &http::Request<B>, res: &Result<http::Response<B2>, E>, retry_backoff: Duration, error_count: u32,
42    ) {
43        let request_uri = SanitizedRequestUri(req.uri());
44        let categorized_error = CategorizedError::try_categorize(res);
45
46        // The HTTP/2 fields are only present when the failure was an HTTP/2 error: `None` values are not recorded.
47        let http2_details = categorized_error.http2_details();
48
49        warn!(
50            error_count,
51            %request_uri,
52            http2.frame = http2_details.map(|details| details.frame.as_str()),
53            http2.reason_code = http2_details.and_then(|details| details.reason).map(u32::from),
54            http2.initiator = http2_details.map(|details| details.initiator.as_str()),
55            "{}. Retrying after {:?}.", categorized_error, retry_backoff
56        );
57
58        if let Some(telemetry) = &self.telemetry {
59            telemetry.increment(categorized_error.retry_cause());
60        }
61    }
62
63    fn after_success(&self, req: &http::Request<B>, _: &Result<http::Response<B2>, E>) {
64        let request_uri = SanitizedRequestUri(req.uri());
65        debug!(%request_uri, "Request succeeded.");
66    }
67}
68
69struct SanitizedRequestUri<'a>(&'a http::Uri);
70
71impl fmt::Display for SanitizedRequestUri<'_> {
72    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
73        // We _should_ always have a scheme and a host, but we'll just make sure they exist first, to be safe. We'll
74        // require needing both to be present to print either of them.
75        let maybe_scheme = self.0.scheme_str();
76        let maybe_host = self.0.host();
77
78        if let (Some(scheme), Some(host)) = (maybe_scheme, maybe_host) {
79            write!(f, "{}://{}", scheme, host)?;
80        }
81
82        let maybe_port = self.0.port_u16();
83        if let Some(port) = maybe_port {
84            write!(f, ":{}", port)?;
85        }
86
87        // Now print the request path, which is always present.
88        write!(f, "{}", self.0.path())
89    }
90}
91
92/// Maximum number of errors visited when walking a source chain.
93///
94/// Chains are short in practice, so the limit only exists to bound the walk if an error ever reports itself, directly or
95/// indirectly, as its own source.
96const MAX_SOURCE_CHAIN_DEPTH: usize = 32;
97
98enum CategorizedError {
99    Client(String),
100    Tls(String),
101    Http2(Http2ErrorDetails),
102    Http(StatusCode),
103    Timeout(TimeoutStage, String),
104    Connection(ConnectionFailure, String),
105    Other(String),
106}
107
108impl CategorizedError {
109    fn try_categorize<B, E>(res: &Result<http::Response<B>, E>) -> Self
110    where
111        E: DynError,
112    {
113        match res {
114            Ok(resp) => Self::Http(resp.status()),
115            Err(e) => Self::extract_nested(e.as_dyn_error()),
116        }
117    }
118
119    fn extract_nested(error: &(dyn std::error::Error + 'static)) -> Self {
120        // Walk the source chain looking for an error we can say something specific about. Wrappers along the way, such
121        // as `hyper-util`'s client error, get no handling of their own: we only care about what they wrap.
122        let mut current = error;
123
124        // The outermost I/O error we can categorize by kind. It's only used if nothing more specific turns up deeper in
125        // the chain, since an I/O error often wraps a more descriptive error that the walk continues into.
126        let mut io_failure = None;
127
128        for _ in 0..MAX_SOURCE_CHAIN_DEPTH {
129            if let Some(rustls_error) = current.downcast_ref::<rustls::Error>() {
130                return Self::from_rustls(rustls_error);
131            }
132
133            if let Some(http2_error) = current.downcast_ref::<h2::Error>() {
134                if let Some(io_error) = http2_error.get_io() {
135                    if let Some(categorized) = Self::from_io(io_error) {
136                        return categorized;
137                    }
138                }
139
140                return Self::Http2(Http2ErrorDetails::from_http2(http2_error));
141            }
142
143            if current.is::<Elapsed>() {
144                return Self::Timeout(TimeoutStage::Request, current.to_string());
145            }
146
147            if io_failure.is_none() {
148                if let Some(io_error) = current.downcast_ref::<io::Error>() {
149                    io_failure = Self::from_io(io_error);
150                }
151            }
152
153            match next_source(current) {
154                Some(source) => current = source,
155                None => break,
156            }
157        }
158
159        if let Some(io_failure) = io_failure {
160            return io_failure;
161        }
162
163        // Nothing in the chain was recognized, so we report the deepest error we reached, since that's the one closest
164        // to the actual failure.
165        if let Some(client_error) = current.downcast_ref::<hyper_util::client::legacy::Error>() {
166            return Self::Client(client_error.to_string());
167        }
168
169        Self::Other(current.to_string())
170    }
171
172    /// Categorizes an I/O error by its kind.
173    ///
174    /// Returns `None` for kinds we have nothing specific to say about, which leaves the error to the generic fallback.
175    fn from_io(error: &io::Error) -> Option<Self> {
176        // Connect and TLS handshake deadlines are surfaced as timed-out I/O errors, so a timeout here is always about
177        // establishing the connection rather than waiting for a response.
178        if error.kind() == io::ErrorKind::TimedOut {
179            return Some(Self::Timeout(TimeoutStage::Connect, error.to_string()));
180        }
181
182        ConnectionFailure::from_io_kind(error.kind()).map(|failure| Self::Connection(failure, error.to_string()))
183    }
184
185    fn http2_details(&self) -> Option<&Http2ErrorDetails> {
186        match self {
187            Self::Http2(details) => Some(details),
188            _ => None,
189        }
190    }
191
192    /// Distills this error into the bounded value that telemetry reports.
193    fn retry_cause(&self) -> RetryCause {
194        match self {
195            Self::Client(_) => RetryCause::Client,
196            Self::Tls(_) => RetryCause::Tls,
197            Self::Http2(details) => RetryCause::Http2 {
198                frame: details.frame,
199                initiator: details.initiator,
200                reason: details.reason.and_then(http2_reason_tag),
201            },
202            Self::Http(_) => RetryCause::HttpStatus,
203            Self::Timeout(stage, _) => RetryCause::Timeout(*stage),
204            Self::Connection(failure, _) => RetryCause::Connection(*failure),
205            Self::Other(_) => RetryCause::Other,
206        }
207    }
208
209    fn from_rustls(error: &rustls::Error) -> Self {
210        // We're really just specializing a few known types of errors to generate a better error message, but otherwise
211        // we'll fallback on the description given by the error itself.
212        let reason = match error {
213            rustls::Error::InvalidCertificate(cert_error) => format!(
214                "peer certificate is invalid: {}",
215                rustls_cert_error_to_string(cert_error)
216            ),
217            _ => error.to_string(),
218        };
219
220        Self::Tls(reason)
221    }
222}
223
224impl fmt::Display for CategorizedError {
225    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
226        match self {
227            CategorizedError::Client(reason) => write!(f, "Request failed due to a client error: {}", reason),
228            CategorizedError::Tls(reason) => write!(f, "Request failed due to a TLS error: {}", reason),
229            CategorizedError::Http2(details) => write!(f, "Request failed due to {}", details),
230            CategorizedError::Http(status_code) => write!(
231                f,
232                "Server responded with non-success status code {}.",
233                status_code.as_str()
234            ),
235            CategorizedError::Timeout(_, reason) => write!(f, "Request failed due to a timeout: {}", reason),
236            CategorizedError::Connection(_, reason) => {
237                write!(f, "Request failed due to a connection error: {}", reason)
238            }
239            CategorizedError::Other(reason) => write!(f, "Request failed: {}", reason),
240        }
241    }
242}
243
244/// Returns the next error in a source chain, if any.
245fn next_source<'a>(error: &'a (dyn std::error::Error + 'static)) -> Option<&'a (dyn std::error::Error + 'static)> {
246    // `std::io::Error::source` skips the error that the I/O error itself wraps, and returns that error's source instead,
247    // so we have to ask for the wrapped error directly.
248    if let Some(io_error) = error.downcast_ref::<std::io::Error>() {
249        if let Some(inner) = io_error.get_ref() {
250            return Some(inner);
251        }
252    }
253
254    error.source()
255}
256
257/// The HTTP/2 frame, if any, that an error came from.
258#[derive(Clone, Copy, Eq, Hash, PartialEq)]
259pub(super) enum Http2Frame {
260    GoAway,
261    Reset,
262    Other,
263}
264
265impl Http2Frame {
266    fn as_str(&self) -> &'static str {
267        match self {
268            Self::GoAway => "go_away",
269            Self::Reset => "reset",
270            Self::Other => "other",
271        }
272    }
273}
274
275/// The side of an HTTP/2 connection that ended the request.
276#[derive(Clone, Copy, Eq, Hash, PartialEq)]
277pub(super) enum Http2Initiator {
278    /// This client: either `h2` enforcing the protocol, or the code driving it.
279    Local,
280
281    /// The remote endpoint, through a GOAWAY or RST_STREAM frame.
282    Remote,
283
284    /// Neither: the error carries no frame to attribute.
285    Unknown,
286}
287
288impl Http2Initiator {
289    fn as_str(&self) -> &'static str {
290        match self {
291            Self::Local => "local",
292            Self::Remote => "remote",
293            Self::Unknown => "unknown",
294        }
295    }
296}
297
298/// The stage of a request that a timeout applies to.
299#[derive(Clone, Copy, Eq, Hash, PartialEq)]
300pub(super) enum TimeoutStage {
301    /// Waiting for a response after the request was sent.
302    Request,
303
304    /// Establishing the connection, including the TLS handshake.
305    Connect,
306}
307
308impl TimeoutStage {
309    fn as_str(&self) -> &'static str {
310        match self {
311            Self::Request => "request",
312            Self::Connect => "connect",
313        }
314    }
315}
316
317/// A transport failure recognized from an I/O error's kind.
318#[derive(Clone, Copy, Eq, Hash, PartialEq)]
319pub(super) enum ConnectionFailure {
320    Refused,
321    Reset,
322    Aborted,
323    BrokenPipe,
324    NotConnected,
325    UnexpectedEof,
326    HostUnreachable,
327    NetworkUnreachable,
328    NetworkDown,
329}
330
331impl ConnectionFailure {
332    fn from_io_kind(kind: io::ErrorKind) -> Option<Self> {
333        match kind {
334            io::ErrorKind::ConnectionRefused => Some(Self::Refused),
335            io::ErrorKind::ConnectionReset => Some(Self::Reset),
336            io::ErrorKind::ConnectionAborted => Some(Self::Aborted),
337            io::ErrorKind::BrokenPipe => Some(Self::BrokenPipe),
338            io::ErrorKind::NotConnected => Some(Self::NotConnected),
339            io::ErrorKind::UnexpectedEof => Some(Self::UnexpectedEof),
340            io::ErrorKind::HostUnreachable => Some(Self::HostUnreachable),
341            io::ErrorKind::NetworkUnreachable => Some(Self::NetworkUnreachable),
342            io::ErrorKind::NetworkDown => Some(Self::NetworkDown),
343            _ => None,
344        }
345    }
346
347    fn as_str(&self) -> &'static str {
348        match self {
349            Self::Refused => "refused",
350            Self::Reset => "reset",
351            Self::Aborted => "aborted",
352            Self::BrokenPipe => "broken_pipe",
353            Self::NotConnected => "not_connected",
354            Self::UnexpectedEof => "unexpected_eof",
355            Self::HostUnreachable => "host_unreachable",
356            Self::NetworkUnreachable => "network_unreachable",
357            Self::NetworkDown => "network_down",
358        }
359    }
360}
361
362/// Returns the tag value for an HTTP/2 reason code, or `None` for a code that `h2` doesn't name.
363///
364/// An unnamed code is left out of the tags rather than turned into one, since the remote endpoint chooses the codes it
365/// sends. The numeric code stays in the retry log.
366fn http2_reason_tag(reason: h2::Reason) -> Option<&'static str> {
367    match reason {
368        h2::Reason::NO_ERROR => Some("no_error"),
369        h2::Reason::PROTOCOL_ERROR => Some("protocol_error"),
370        h2::Reason::INTERNAL_ERROR => Some("internal_error"),
371        h2::Reason::FLOW_CONTROL_ERROR => Some("flow_control_error"),
372        h2::Reason::SETTINGS_TIMEOUT => Some("settings_timeout"),
373        h2::Reason::STREAM_CLOSED => Some("stream_closed"),
374        h2::Reason::FRAME_SIZE_ERROR => Some("frame_size_error"),
375        h2::Reason::REFUSED_STREAM => Some("refused_stream"),
376        h2::Reason::CANCEL => Some("cancel"),
377        h2::Reason::COMPRESSION_ERROR => Some("compression_error"),
378        h2::Reason::CONNECT_ERROR => Some("connect_error"),
379        h2::Reason::ENHANCE_YOUR_CALM => Some("enhance_your_calm"),
380        h2::Reason::INADEQUATE_SECURITY => Some("inadequate_security"),
381        h2::Reason::HTTP_1_1_REQUIRED => Some("http_1_1_required"),
382        _ => None,
383    }
384}
385
386const TAG_CAUSE: &str = "cause";
387const TAG_FRAME: &str = "frame";
388const TAG_INITIATOR: &str = "initiator";
389const TAG_REASON: &str = "reason";
390
391/// The bounded classification of a retry, as reported to telemetry.
392///
393/// Every part of this value is an enumerated variant or a fixed string, so the set of tag combinations it can produce is
394/// known at compile time. That's what bounds the cardinality of the retry cause metric, and it's why error text, GOAWAY
395/// debug data, and unrecognized reason codes stay in the retry log instead.
396#[derive(Clone, Copy, Eq, Hash, PartialEq)]
397pub(super) enum RetryCause {
398    HttpStatus,
399    Http2 {
400        frame: Http2Frame,
401        initiator: Http2Initiator,
402        reason: Option<&'static str>,
403    },
404    Tls,
405    Timeout(TimeoutStage),
406    Connection(ConnectionFailure),
407    Client,
408    Other,
409}
410
411impl RetryCause {
412    /// Returns the tags for this cause's counter.
413    pub(super) fn tags(&self) -> Vec<(&'static str, &'static str)> {
414        match self {
415            Self::HttpStatus => vec![(TAG_CAUSE, "http_status")],
416            Self::Http2 {
417                frame,
418                initiator,
419                reason,
420            } => {
421                let mut tags = vec![
422                    (TAG_CAUSE, "http2"),
423                    (TAG_FRAME, frame.as_str()),
424                    (TAG_INITIATOR, initiator.as_str()),
425                ];
426
427                if let Some(reason) = reason {
428                    tags.push((TAG_REASON, reason));
429                }
430
431                tags
432            }
433            Self::Tls => vec![(TAG_CAUSE, "tls")],
434            Self::Timeout(stage) => vec![(TAG_CAUSE, "timeout"), (TAG_REASON, stage.as_str())],
435            Self::Connection(failure) => vec![(TAG_CAUSE, "connection"), (TAG_REASON, failure.as_str())],
436            Self::Client => vec![(TAG_CAUSE, "client")],
437            Self::Other => vec![(TAG_CAUSE, "other")],
438        }
439    }
440}
441
442/// The details of an `h2::Error` that we report.
443///
444/// We snapshot a small set of fields instead of holding the error, so that we never log its `Debug` output or the debug
445/// data carried by a GOAWAY frame.
446struct Http2ErrorDetails {
447    frame: Http2Frame,
448    initiator: Http2Initiator,
449    reason: Option<h2::Reason>,
450
451    /// The error's own text, used only when there is no reason code to describe the failure.
452    fallback: Option<String>,
453}
454
455impl Http2ErrorDetails {
456    fn from_http2(error: &h2::Error) -> Self {
457        let frame = if error.is_go_away() {
458            Http2Frame::GoAway
459        } else if error.is_reset() {
460            Http2Frame::Reset
461        } else {
462            Http2Frame::Other
463        };
464
465        // Only GOAWAY and RST_STREAM errors have a side that sent them. `h2` distinguishes the errors it raises itself
466        // from the ones the calling code causes, but both mean the same thing here: the request ended locally.
467        let initiator = match frame {
468            Http2Frame::GoAway | Http2Frame::Reset if error.is_remote() => Http2Initiator::Remote,
469            Http2Frame::GoAway | Http2Frame::Reset => Http2Initiator::Local,
470            Http2Frame::Other => Http2Initiator::Unknown,
471        };
472
473        // The error's text includes GOAWAY debug data, so we only take it when we have nothing better.
474        let reason = error.reason();
475        let fallback = reason.is_none().then(|| error.to_string());
476
477        Self {
478            frame,
479            initiator,
480            reason,
481            fallback,
482        }
483    }
484}
485
486impl fmt::Display for Http2ErrorDetails {
487    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
488        let description = match self.frame {
489            Http2Frame::GoAway => "an HTTP/2 GOAWAY",
490            Http2Frame::Reset => "an HTTP/2 stream reset",
491            Http2Frame::Other => "an HTTP/2 error",
492        };
493
494        write!(f, "{}", description)?;
495
496        match self.initiator {
497            Http2Initiator::Local => write!(f, " sent by this client")?,
498            Http2Initiator::Remote => write!(f, " received from the remote peer")?,
499            Http2Initiator::Unknown => {}
500        }
501
502        match self.reason {
503            Some(reason) => write!(f, " (reason: {}, code {})", reason.description(), u32::from(reason)),
504            None => match &self.fallback {
505                Some(fallback) => write!(f, ": {}", fallback),
506                None => Ok(()),
507            },
508        }
509    }
510}
511
512fn rustls_cert_error_to_string(cert_error: &rustls::CertificateError) -> Cow<'static, str> {
513    match cert_error {
514        rustls::CertificateError::BadEncoding => "certificate incorrectly encoded".into(),
515        rustls::CertificateError::Expired => "certificate expired (current time is after notAfter time)".into(),
516        rustls::CertificateError::NotValidYet => {
517            "certificate not valid yet (current time is before notBefore time)".into()
518        }
519        rustls::CertificateError::Revoked => "certificate has been revoked".into(),
520        rustls::CertificateError::UnhandledCriticalExtension => {
521            "certificate contains an extension marked critical, but it was not processed by the certificate validator"
522                .into()
523        }
524        rustls::CertificateError::UnknownIssuer => "certificate chain is not issued by a known root certificate".into(),
525        rustls::CertificateError::UnknownRevocationStatus => {
526            "certificate's revocation status could not be determined".into()
527        }
528        rustls::CertificateError::ExpiredRevocationList => {
529            "certificate's revocation status could not be determined due to an expired CRL".into()
530        }
531        rustls::CertificateError::BadSignature => {
532            "certificate is not signed correctly by the key of its alleged issuer".into()
533        }
534        rustls::CertificateError::NotValidForName => "certificate is not valid for the given entity name".into(),
535        rustls::CertificateError::InvalidPurpose => "certificate is not valid for the requested purpose".into(),
536        rustls::CertificateError::ApplicationVerificationFailure => {
537            "certificate is valid overall, but the handshake was rejected".into()
538        }
539
540        // This one could be a generic error that doesn't fit the above, returned by `rustls`, or it could be coming from
541        // a custom certificate verifier which we don't know about, or can't reasonably know about to compensate for
542        // here... so we'll just return it as-is.
543        rustls::CertificateError::Other(other) => format!("generic error: {}", other).into(),
544        other => format!("generic unhandled error: {:?}", other).into(),
545    }
546}
547
548// Market trait for accepting generically-typed errors that can be downcasted to dynamically-dispatched trait references.
549trait DynError {
550    fn as_dyn_error(&self) -> &(dyn std::error::Error + 'static);
551}
552
553impl DynError for Box<dyn std::error::Error + Send + Sync> {
554    fn as_dyn_error(&self) -> &(dyn std::error::Error + 'static) {
555        &**self
556    }
557}
558
559#[cfg(test)]
560mod tests {
561    use std::{collections::HashMap, future::Future, io};
562
563    use bytes::Bytes;
564    use http::{Response, StatusCode, Uri};
565    use metrics::{Key, Label, SharedString, Unit};
566    use metrics_util::{
567        debugging::{DebugValue, DebuggingRecorder},
568        CompositeKey, MetricKind,
569    };
570    use saluki_metrics::MetricsBuilder;
571
572    use super::*;
573
574    type BoxError = Box<dyn std::error::Error + Send + Sync>;
575    type MetricsSnapshot = HashMap<CompositeKey, (Option<Unit>, Option<SharedString>, DebugValue)>;
576
577    const RETRY_CAUSES_TOTAL: &str = "network_http_requests_retry_causes_total";
578
579    /// Enough empty DATA frames to exhaust `h2`'s budget for them, which trips its abuse protection.
580    const EMPTY_DATA_FRAMES: usize = 256;
581
582    /// An error that reports another error as its source, for building nested chains.
583    #[derive(Debug)]
584    struct NestedError {
585        message: &'static str,
586        source: BoxError,
587    }
588
589    impl NestedError {
590        fn new(message: &'static str, source: impl Into<BoxError>) -> Self {
591            Self {
592                message,
593                source: source.into(),
594            }
595        }
596    }
597
598    impl fmt::Display for NestedError {
599        fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
600            write!(f, "{}", self.message)
601        }
602    }
603
604    impl std::error::Error for NestedError {
605        fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
606            Some(&*self.source)
607        }
608    }
609
610    fn categorize(res: Result<Response<()>, BoxError>) -> String {
611        CategorizedError::try_categorize(&res).to_string()
612    }
613
614    fn categorize_error(error: impl Into<BoxError>) -> CategorizedError {
615        let res: Result<Response<()>, BoxError> = Err(error.into());
616        CategorizedError::try_categorize(&res)
617    }
618
619    /// Returns the telemetry tags for a failed request, which is what bounds the cause's cardinality.
620    fn cause_tags(error: impl Into<BoxError>) -> Vec<(&'static str, &'static str)> {
621        categorize_error(error).retry_cause().tags()
622    }
623
624    /// Runs a request over an in-memory HTTP/2 connection and returns the error the client sees.
625    ///
626    /// The server side completes the handshake, waits until the client has sent its request, and then runs `reject`,
627    /// which is expected to fail either the stream or the connection. `reject` hands the connection back, since the
628    /// server has to keep polling it to flush what `reject` queued, and has to hold it open so that closing the socket
629    /// doesn't race the client's read.
630    async fn failed_http2_request<F, Fut>(reject: F) -> h2::Error
631    where
632        F: FnOnce(h2::server::Connection<tokio::io::DuplexStream, Bytes>) -> Fut + Send + 'static,
633        Fut: Future<Output = h2::server::Connection<tokio::io::DuplexStream, Bytes>> + Send,
634    {
635        let (client_io, server_io) = tokio::io::duplex(4096);
636        let (request_sent_tx, request_sent_rx) = tokio::sync::oneshot::channel();
637
638        tokio::spawn(async move {
639            let connection = h2::server::handshake(server_io).await.unwrap();
640            request_sent_rx.await.unwrap();
641
642            let mut connection = reject(connection).await;
643            let _ = std::future::poll_fn(|cx| connection.poll_closed(cx)).await;
644            std::future::pending::<()>().await;
645        });
646
647        let (send_request, connection) = h2::client::handshake(client_io).await.unwrap();
648        let driver = tokio::spawn(connection);
649
650        let mut send_request = send_request.ready().await.unwrap();
651        let request = http::Request::get("https://localhost/api/v2/series").body(()).unwrap();
652        let (response, _send_stream) = send_request.send_request(request, true).unwrap();
653        request_sent_tx.send(()).unwrap();
654
655        let error = response.await.expect_err("request should have failed");
656        driver.abort();
657
658        error
659    }
660
661    /// Runs a request over an in-memory HTTP/2 connection whose server closes without responding.
662    async fn failed_http2_io_request() -> h2::Error {
663        let (client_io, server_io) = tokio::io::duplex(4096);
664
665        tokio::spawn(async move {
666            let mut connection = h2::server::handshake(server_io).await.unwrap();
667            let _ = connection.accept().await;
668        });
669
670        let (send_request, connection) = h2::client::handshake(client_io).await.unwrap();
671        let driver = tokio::spawn(connection);
672
673        let mut send_request = send_request.ready().await.unwrap();
674        let request = http::Request::get("https://localhost/api/v2/series").body(()).unwrap();
675        let (response, _send_stream) = send_request.send_request(request, true).unwrap();
676        let error = response
677            .await
678            .expect_err("request should fail when the connection closes");
679        driver.abort();
680
681        error
682    }
683
684    /// Runs a request over an in-memory HTTP/2 connection whose server floods the response body with empty DATA frames,
685    /// and returns the error our own `h2` raises when its abuse protection trips.
686    ///
687    /// This is the failure that ended requests during the retry storm, apart from which frames `h2` charged against its
688    /// budget. As in [`failed_http2_request`], the server holds the connection open afterwards so that closing the
689    /// socket doesn't race the client's read.
690    async fn http2_abuse_protection_error() -> h2::Error {
691        let (client_io, server_io) = tokio::io::duplex(64 * 1024);
692
693        tokio::spawn(async move {
694            let mut connection = h2::server::handshake(server_io).await.unwrap();
695            let (_request, mut respond) = connection.accept().await.unwrap().unwrap();
696            let mut send_stream = respond.send_response(Response::new(()), false).unwrap();
697            for _ in 0..EMPTY_DATA_FRAMES {
698                send_stream.send_data(Bytes::new(), false).unwrap();
699            }
700
701            let _ = std::future::poll_fn(|cx| connection.poll_closed(cx)).await;
702            std::future::pending::<()>().await;
703        });
704
705        let (send_request, connection) = h2::client::handshake(client_io).await.unwrap();
706        let driver = tokio::spawn(connection);
707
708        let mut send_request = send_request.ready().await.unwrap();
709        let request = http::Request::get("https://localhost/api/v2/series").body(()).unwrap();
710        let (response, _send_stream) = send_request.send_request(request, true).unwrap();
711
712        let mut body = response.await.expect("response head should arrive").into_body();
713        let error = loop {
714            match body.data().await {
715                Some(Ok(_)) => continue,
716                Some(Err(error)) => break error,
717                None => panic!("response body ended without an error"),
718            }
719        };
720        driver.abort();
721
722        error
723    }
724
725    /// Drives one retry decision through the lifecycle, which is what logs and counts it.
726    fn record_retry(lifecycle: &StandardHttpRetryLifecycle, res: Result<Response<()>, BoxError>) {
727        let request = http::Request::get("https://example.com/api/v2/series")
728            .body(())
729            .unwrap();
730        lifecycle.before_retry(&request, &res, Duration::from_millis(1), 1);
731    }
732
733    /// Returns the value of the retry cause counter carrying `tags`, or panics if it was never registered.
734    ///
735    /// A snapshot reports each counter's value once, so take one snapshot and read every counter from it.
736    #[track_caller]
737    fn retry_cause_count(snapshot: &MetricsSnapshot, tags: &[(&'static str, &'static str)]) -> u64 {
738        let mut labels = vec![Label::new("domain", "https://example.com")];
739        labels.extend(tags.iter().map(|(name, value)| Label::new(*name, *value)));
740
741        let key = CompositeKey::new(MetricKind::Counter, Key::from_parts(RETRY_CAUSES_TOTAL, labels));
742        match snapshot.get(&key) {
743            Some((_, _, DebugValue::Counter(value))) => *value,
744            _ => panic!("no retry cause counter for {:?}", tags),
745        }
746    }
747
748    fn error_response(status: StatusCode) -> Result<Response<()>, BoxError> {
749        Ok(Response::builder().status(status).body(()).unwrap())
750    }
751
752    #[test]
753    fn categorizes_http_status_response() {
754        // A response (any `Ok`) is categorized by its status code, and non-success codes render with the code value. The
755        // code itself isn't tagged: `network_http_requests_errors_total` already counts responses by status code.
756        let response = Response::builder()
757            .status(StatusCode::INTERNAL_SERVER_ERROR)
758            .body(())
759            .unwrap();
760        let res: Result<Response<()>, BoxError> = Ok(response);
761        let categorized = CategorizedError::try_categorize(&res);
762
763        assert_eq!(
764            categorized.to_string(),
765            "Server responded with non-success status code 500."
766        );
767        assert_eq!(categorized.retry_cause().tags(), vec![("cause", "http_status")]);
768    }
769
770    #[test]
771    fn categorizes_io_error_kinds_as_connection_failures() {
772        // Transport failures we can name from the I/O error kind get their own reason, so they don't share a bucket with
773        // protocol errors.
774        let cases = [
775            (io::ErrorKind::ConnectionRefused, "refused", "connection refused"),
776            (io::ErrorKind::ConnectionReset, "reset", "connection reset"),
777            (io::ErrorKind::ConnectionAborted, "aborted", "connection aborted"),
778            (io::ErrorKind::BrokenPipe, "broken_pipe", "broken pipe"),
779            (io::ErrorKind::NotConnected, "not_connected", "not connected"),
780            (io::ErrorKind::UnexpectedEof, "unexpected_eof", "unexpected end of file"),
781            (io::ErrorKind::HostUnreachable, "host_unreachable", "host unreachable"),
782            (
783                io::ErrorKind::NetworkUnreachable,
784                "network_unreachable",
785                "network unreachable",
786            ),
787            (io::ErrorKind::NetworkDown, "network_down", "network down"),
788        ];
789
790        for (kind, expected_reason, expected_message) in cases {
791            let err: BoxError = Box::new(io::Error::from(kind));
792            assert_eq!(
793                categorize(Err(err)),
794                format!("Request failed due to a connection error: {}", expected_message)
795            );
796
797            let err: BoxError = Box::new(io::Error::from(kind));
798            assert_eq!(
799                cause_tags(err),
800                vec![("cause", "connection"), ("reason", expected_reason)],
801                "{:?} should report a connection failure",
802                kind
803            );
804        }
805    }
806
807    #[test]
808    fn categorizes_request_timeout_as_timeout() {
809        // The client's per-request timeout fires above the transport, so nothing in the chain says what failed beyond
810        // the elapsed deadline itself.
811        let err: BoxError = Box::new(Elapsed::new());
812        assert_eq!(
813            categorize(Err(err)),
814            "Request failed due to a timeout: request timed out"
815        );
816
817        let err: BoxError = Box::new(Elapsed::new());
818        assert_eq!(cause_tags(err), vec![("cause", "timeout"), ("reason", "request")]);
819    }
820
821    #[test]
822    fn categorizes_timed_out_io_error_as_connect_timeout() {
823        // Connect and TLS handshake deadlines reach us as timed-out I/O errors, and their message says which one it was.
824        let inner = NestedError::new(
825            "connecting to endpoint",
826            io::Error::new(io::ErrorKind::TimedOut, "TLS handshake timed out"),
827        );
828        let err: BoxError = Box::new(NestedError::new("sending request", inner));
829        assert_eq!(
830            categorize(Err(err)),
831            "Request failed due to a timeout: TLS handshake timed out"
832        );
833
834        let err: BoxError = Box::new(io::Error::from(io::ErrorKind::TimedOut));
835        assert_eq!(cause_tags(err), vec![("cause", "timeout"), ("reason", "connect")]);
836    }
837
838    #[test]
839    fn categorizes_rustls_certificate_error_as_tls() {
840        // A rustls certificate error is specialized into a TLS category with a human-readable reason. The reason stays in
841        // the log: rustls has far too many error variants to tag.
842        let err: BoxError = Box::new(rustls::Error::InvalidCertificate(rustls::CertificateError::Expired));
843        assert_eq!(
844            categorize(Err(err)),
845            "Request failed due to a TLS error: peer certificate is invalid: certificate expired (current time is after notAfter time)"
846        );
847
848        let err: BoxError = Box::new(rustls::Error::InvalidCertificate(rustls::CertificateError::Expired));
849        assert_eq!(cause_tags(err), vec![("cause", "tls")]);
850    }
851
852    #[test]
853    fn categorizes_io_wrapped_rustls_error_by_unwrapping_source() {
854        // An io::Error that wraps a rustls error is unwrapped via its source and categorized as TLS, not reported as
855        // a generic io failure.
856        let inner = rustls::Error::InvalidCertificate(rustls::CertificateError::Revoked);
857        let err: BoxError = Box::new(io::Error::other(inner));
858        assert_eq!(
859            categorize(Err(err)),
860            "Request failed due to a TLS error: peer certificate is invalid: certificate has been revoked"
861        );
862
863        // A recognizable I/O error kind doesn't win over the error it wraps, since the TLS failure is the more useful of
864        // the two.
865        let inner = rustls::Error::InvalidCertificate(rustls::CertificateError::UnknownIssuer);
866        let err: BoxError = Box::new(io::Error::new(io::ErrorKind::ConnectionReset, inner));
867        assert_eq!(cause_tags(err), vec![("cause", "tls")]);
868    }
869
870    #[test]
871    fn categorizes_deepest_source_when_nothing_is_recognized() {
872        // With no recognized error in the chain, the deepest source is reported, since it sits closest to the failure.
873        let inner = NestedError::new("connecting to endpoint", io::Error::other("something went sideways"));
874        let err: BoxError = Box::new(NestedError::new("sending request", inner));
875        assert_eq!(categorize(Err(err)), "Request failed: something went sideways");
876
877        let inner = NestedError::new("connecting to endpoint", io::Error::other("something went sideways"));
878        let err: BoxError = Box::new(NestedError::new("sending request", inner));
879        assert_eq!(cause_tags(err), vec![("cause", "other")]);
880    }
881
882    #[test]
883    fn categorizes_nested_http2_error_as_http2() {
884        // An HTTP/2 error is found however deeply it is wrapped, and the wrappers add nothing to the message.
885        let inner = NestedError::new(
886            "connection closed",
887            io::Error::other(h2::Error::from(h2::Reason::ENHANCE_YOUR_CALM)),
888        );
889        let err: BoxError = Box::new(NestedError::new("sending request", inner));
890
891        let categorized = categorize_error(err);
892        assert_eq!(
893            categorized.to_string(),
894            "Request failed due to an HTTP/2 error (reason: detected excessive load generating behavior, code 11)"
895        );
896
897        let details = categorized.http2_details().expect("should be an HTTP/2 error");
898        assert_eq!(details.reason.map(u32::from), Some(11));
899        assert_eq!(
900            categorized.retry_cause().tags(),
901            vec![
902                ("cause", "http2"),
903                ("frame", "other"),
904                ("initiator", "unknown"),
905                ("reason", "enhance_your_calm")
906            ]
907        );
908    }
909
910    #[tokio::test]
911    async fn categorizes_http2_io_error_by_its_io_kind() {
912        // A transport failure wrapped by `h2` is still a connection failure; the HTTP/2 layer adds no useful protocol
913        // details in this case.
914        let error = failed_http2_io_request().await;
915        assert!(error.get_io().is_some(), "the h2 error should wrap the I/O failure");
916        assert_eq!(
917            cause_tags(error),
918            vec![("cause", "connection"), ("reason", "broken_pipe")]
919        );
920    }
921
922    #[test]
923    fn categorizes_http2_reason_only_error_as_http2() {
924        // An error carrying only a reason code has no frame behind it, so neither side of the connection can be held
925        // responsible for it.
926        let categorized = categorize_error(h2::Error::from(h2::Reason::INTERNAL_ERROR));
927        assert_eq!(
928            categorized.to_string(),
929            "Request failed due to an HTTP/2 error (reason: unexpected internal error encountered, code 2)"
930        );
931        assert_eq!(
932            categorized.retry_cause().tags(),
933            vec![
934                ("cause", "http2"),
935                ("frame", "other"),
936                ("initiator", "unknown"),
937                ("reason", "internal_error")
938            ]
939        );
940    }
941
942    #[test]
943    fn categorizes_http2_unknown_reason_code_as_http2() {
944        // Reason codes that `h2` has no description for are still reported by their numeric value, but they don't become
945        // tag values: the peer, not us, decides what codes it sends.
946        let categorized = categorize_error(h2::Error::from(h2::Reason::from(9_001)));
947        assert_eq!(
948            categorized.to_string(),
949            "Request failed due to an HTTP/2 error (reason: unknown reason, code 9001)"
950        );
951
952        let details = categorized.http2_details().expect("should be an HTTP/2 error");
953        assert_eq!(details.reason.map(u32::from), Some(9_001));
954        assert_eq!(
955            categorized.retry_cause().tags(),
956            vec![("cause", "http2"), ("frame", "other"), ("initiator", "unknown")]
957        );
958    }
959
960    #[test]
961    fn http2_reason_codes_are_named_for_every_code_h2_defines() {
962        // Each reason code `h2` names gets a tag value, so a retry can be attributed to a specific protocol failure.
963        let codes = [
964            h2::Reason::NO_ERROR,
965            h2::Reason::PROTOCOL_ERROR,
966            h2::Reason::INTERNAL_ERROR,
967            h2::Reason::FLOW_CONTROL_ERROR,
968            h2::Reason::SETTINGS_TIMEOUT,
969            h2::Reason::STREAM_CLOSED,
970            h2::Reason::FRAME_SIZE_ERROR,
971            h2::Reason::REFUSED_STREAM,
972            h2::Reason::CANCEL,
973            h2::Reason::COMPRESSION_ERROR,
974            h2::Reason::CONNECT_ERROR,
975            h2::Reason::ENHANCE_YOUR_CALM,
976            h2::Reason::INADEQUATE_SECURITY,
977            h2::Reason::HTTP_1_1_REQUIRED,
978        ];
979
980        let mut named = Vec::new();
981        for code in codes {
982            let tag = http2_reason_tag(code).unwrap_or_else(|| panic!("code {} should be named", u32::from(code)));
983            named.push(tag);
984        }
985
986        named.sort_unstable();
987        named.dedup();
988        assert_eq!(named.len(), codes.len(), "each code should have its own name");
989    }
990
991    #[tokio::test]
992    async fn categorizes_remote_http2_goaway_as_http2() {
993        // A GOAWAY from the peer is reported as such, along with the fact that it came from the remote.
994        let error = failed_http2_request(|mut connection| async move {
995            connection.abrupt_shutdown(h2::Reason::ENHANCE_YOUR_CALM);
996            connection
997        })
998        .await;
999
1000        let categorized = categorize_error(io::Error::other(error));
1001        assert_eq!(
1002            categorized.to_string(),
1003            "Request failed due to an HTTP/2 GOAWAY received from the remote peer (reason: detected excessive load generating behavior, code 11)"
1004        );
1005
1006        assert_eq!(
1007            categorized.retry_cause().tags(),
1008            vec![
1009                ("cause", "http2"),
1010                ("frame", "go_away"),
1011                ("initiator", "remote"),
1012                ("reason", "enhance_your_calm")
1013            ]
1014        );
1015    }
1016
1017    #[tokio::test]
1018    async fn categorizes_local_http2_goaway_as_locally_initiated() {
1019        // Regression coverage for the retry storm that motivated this: `h2`'s abuse protection counted legitimate small
1020        // DATA frames, so our own client sent GOAWAY(ENHANCE_YOUR_CALM) and every in-flight request was retried. The
1021        // retry looked identical to one caused by the remote endpoint, which is what reporting the initiator fixes.
1022        let error = http2_abuse_protection_error().await;
1023
1024        let categorized = categorize_error(io::Error::other(error));
1025        assert_eq!(
1026            categorized.to_string(),
1027            "Request failed due to an HTTP/2 GOAWAY sent by this client (reason: detected excessive load generating behavior, code 11)"
1028        );
1029        assert_eq!(
1030            categorized.retry_cause().tags(),
1031            vec![
1032                ("cause", "http2"),
1033                ("frame", "go_away"),
1034                ("initiator", "local"),
1035                ("reason", "enhance_your_calm")
1036            ]
1037        );
1038    }
1039
1040    #[tokio::test]
1041    async fn categorizes_remote_http2_reset_as_http2() {
1042        // A stream reset from the peer is reported as a stream-level failure that came from the remote.
1043        let error = failed_http2_request(|mut connection| async move {
1044            let (_request, mut respond) = connection.accept().await.unwrap().unwrap();
1045            respond.send_reset(h2::Reason::REFUSED_STREAM);
1046            connection
1047        })
1048        .await;
1049
1050        let categorized = categorize_error(io::Error::other(error));
1051        assert_eq!(
1052            categorized.to_string(),
1053            "Request failed due to an HTTP/2 stream reset received from the remote peer (reason: refused stream before processing any application logic, code 7)"
1054        );
1055
1056        assert_eq!(
1057            categorized.retry_cause().tags(),
1058            vec![
1059                ("cause", "http2"),
1060                ("frame", "reset"),
1061                ("initiator", "remote"),
1062                ("reason", "refused_stream")
1063            ]
1064        );
1065    }
1066
1067    #[test]
1068    fn retries_are_counted_by_cause() {
1069        // Before the cause tags existed, these three retries were indistinguishable in telemetry: the status responses
1070        // and the timeout landed in the same broad transaction-error bucket.
1071        let recorder = DebuggingRecorder::new();
1072        let snapshotter = recorder.snapshotter();
1073        let telemetry = RetryCauseTelemetry::from_builder(&MetricsBuilder::default(), "https://example.com");
1074        let lifecycle = StandardHttpRetryLifecycle::new().with_telemetry(telemetry);
1075
1076        metrics::with_local_recorder(&recorder, || {
1077            record_retry(&lifecycle, error_response(StatusCode::SERVICE_UNAVAILABLE));
1078            record_retry(&lifecycle, error_response(StatusCode::INTERNAL_SERVER_ERROR));
1079            record_retry(&lifecycle, Err(Box::new(Elapsed::new())));
1080        });
1081
1082        let snapshot = snapshotter.snapshot().into_hashmap();
1083
1084        // Both status codes share one series, so a retry storm's shape doesn't depend on how many codes it spans.
1085        assert_eq!(retry_cause_count(&snapshot, &[("cause", "http_status")]), 2);
1086        assert_eq!(
1087            retry_cause_count(&snapshot, &[("cause", "timeout"), ("reason", "request")]),
1088            1
1089        );
1090
1091        let series = snapshot
1092            .keys()
1093            .filter(|key| key.key().name() == RETRY_CAUSES_TOTAL)
1094            .count();
1095        assert_eq!(series, 2);
1096    }
1097
1098    #[tokio::test]
1099    async fn locally_initiated_http2_retries_are_counted_as_local() {
1100        // The signal the retry storm investigation lacked: whether this client or the remote endpoint ended the request.
1101        let error = http2_abuse_protection_error().await;
1102
1103        let recorder = DebuggingRecorder::new();
1104        let snapshotter = recorder.snapshotter();
1105        let telemetry = RetryCauseTelemetry::from_builder(&MetricsBuilder::default(), "https://example.com");
1106        let lifecycle = StandardHttpRetryLifecycle::new().with_telemetry(telemetry);
1107
1108        metrics::with_local_recorder(&recorder, || {
1109            record_retry(&lifecycle, Err(Box::new(io::Error::other(error))));
1110        });
1111
1112        assert_eq!(
1113            retry_cause_count(
1114                &snapshotter.snapshot().into_hashmap(),
1115                &[
1116                    ("cause", "http2"),
1117                    ("frame", "go_away"),
1118                    ("initiator", "local"),
1119                    ("reason", "enhance_your_calm")
1120                ]
1121            ),
1122            1
1123        );
1124    }
1125
1126    #[test]
1127    fn retries_are_not_counted_without_telemetry() {
1128        // Retry logging works on its own, so a policy built without telemetry registers nothing.
1129        let recorder = DebuggingRecorder::new();
1130        let snapshotter = recorder.snapshotter();
1131        let lifecycle = StandardHttpRetryLifecycle::new();
1132
1133        metrics::with_local_recorder(&recorder, || {
1134            record_retry(&lifecycle, Err(Box::new(Elapsed::new())));
1135        });
1136
1137        assert!(snapshotter.snapshot().into_hashmap().is_empty());
1138    }
1139
1140    #[test]
1141    fn sanitized_request_uri_display() {
1142        // Scheme, host, explicit port, and path are all rendered.
1143        let uri: Uri = "http://localhost:8125/foo/bar".parse().unwrap();
1144        assert_eq!(SanitizedRequestUri(&uri).to_string(), "http://localhost:8125/foo/bar");
1145
1146        // A default (implicit) port is omitted.
1147        let uri: Uri = "https://api.datadoghq.com/api/v1/series".parse().unwrap();
1148        assert_eq!(
1149            SanitizedRequestUri(&uri).to_string(),
1150            "https://api.datadoghq.com/api/v1/series"
1151        );
1152
1153        // With neither scheme nor host, only the path is rendered.
1154        let uri: Uri = "/health".parse().unwrap();
1155        assert_eq!(SanitizedRequestUri(&uri).to_string(), "/health");
1156    }
1157}
saluki_io/net/util/retry/lifecycle/http.rs

saluki_io/net/util/retry/lifecycle/
http.rs
