fix(app/core): do not immediately retry negative ttl errors (#4450)

`linkerd_app_core::control` provides utilities used by the data plane to
communicate with the linkerd control plane. this includes, among other
features such as load-balancing and configurability for settings like
connection timeout durations, an error recovery that respects DNS
record's negative TTL.

as of today, we do this within an inline, anonymous closure.

this commit pulls this business logic out of an inline closure, and into
an explicit pair of structures.

ResolveRecover is the Recover implementation that handles identifying
the proper backoff strategy, when presented with a given boxed error.
ResolveBackoff is the structure that acts as the sum type that
encompasses either a TTL-driven interval, or an exponential backoff.

see also, #4449. that introduces some additional
guardrails to prevent panicking if a negative ttl of zero is
encountered.

as part of this code motion, this commit inserts a call to
`tokio::time::Interval::reset()` to the `Recover` implementation that
extracts negative TTL's from `hickory_resolver` errors.

this means that, upon resolution errors with a negative TTL, we will no
longer immediately retry, and instead wait for the prescribed time
before attempting once more.

introducing test coverage for this is difficult because we cannot create
a `ResolveError` ourselves, and introducing e.g. a trait to inject here
would incur an excessive amount of boilerplate and complexity.

to provide assurance that this is correct, see this small playground
example, in which we poll an `Interval` with and without this call to
`reset()`. note that when calling reset, it will no longer immediately
return `Poll::Ready(_)` upon the first call to `tick()`.

```rust
 #[tokio::main]
 async fn main() {
     let duration = std::time::Duration::from_secs(1);
     let mut interval = tokio::time::interval(duration);
     // interval.reset();

     let start = std::time::Instant::now();
     for i in 1..5 {
         interval.tick().await;
         let elapsed = start.elapsed().as_millis();
         println!("#{i} - {elapsed}ms")
     }
 }
```

```
 ; cargo run
 #1 - 1ms
 #2 - 1001ms
 #3 - 2001ms
 #4 - 3001ms
```

with a reset, to avoid first poll being ready:

```rust
 #[tokio::main]
 async fn main() {
     let duration = std::time::Duration::from_secs(1);
     let mut interval = tokio::time::interval(duration);
     interval.reset();

     let start = std::time::Instant::now();
     for i in 1..5 {
         interval.tick().await;
         let elapsed = start.elapsed().as_millis();
         println!("#{i} - {elapsed}ms")
     }
 }
```

```
 ; cargo run
 #1 - 1001ms
 #2 - 2001ms
 #3 - 3001ms
 #4 - 4001ms
```

Signed-off-by: katelyn martin <kate@buoyant.io>
Co-authored-by: Alejandro Martinez Ruiz <alex@flawedcode.org>

Author: cratelyn

linkerd/app/core/src/control.rs

@@ -2,12 +2,10 @@ use crate::{
     classify, config, dns, identity, metrics, proxy::http, svc, tls, transport::ConnectTcp, Addr,
     Error,
 };
-use futures::future::Either;
 use linkerd_metrics::prom;
 use std::fmt;
 use tokio::time;
-use tokio_stream::{wrappers::IntervalStream, StreamExt};
-use tracing::{info_span, warn};
+use tracing::info_span;
 
 #[derive(Clone, Debug)]
 pub struct Config {
@@ -106,26 +104,6 @@ impl Config {
         let addr = self.addr;
         tracing::trace!(%addr, "Building");
 
-        // When a DNS resolution fails, log the error and use the TTL, if there
-        // is one, to drive re-resolution attempts.
-        let resolve_backoff = {
-            let backoff = self.connect.backoff;
-            move |error: Error| {
-                warn!(error, "Failed to resolve control-plane component");
-                if let Some(e) = crate::errors::cause_ref::<dns::ResolveError>(&*error) {
-                    if let Some(ttl) = e.negative_ttl() {
-                        return Ok::<_, Error>(Either::Left(
-                            IntervalStream::new(time::interval(ttl)).map(|_| ()),
-                        ));
-                    }
-                }
-
-                // If the error didn't give us a TTL, use the default jittered
-                // backoff.
-                Ok(Either::Right(backoff.stream()))
-            }
-        };
-
         let client = svc::stack(ConnectTcp::new(
             self.connect.keepalive,
             self.connect.user_timeout,
@@ -151,7 +129,11 @@ impl Config {
 
         let balance = endpoint
             .lift_new()
-            .push(self::balance::layer(metrics.balance, dns, resolve_backoff))
+            .push(self::balance::layer(
+                metrics.balance,
+                dns,
+                self.connect.backoff,
+            ))
             .push(legacy_metrics.to_layer::<classify::Response, _, _>())
             .push(classify::NewClassify::layer_default());
 
@@ -251,20 +233,24 @@ mod balance {
         proxy::{dns_resolve::DnsResolve, http, resolve::recover},
         svc, tls,
     };
+    use futures::Stream;
+    use linkerd_error::Recover;
+    use linkerd_exp_backoff::{ExponentialBackoff, ExponentialBackoffStream};
     use linkerd_stack::ExtractParam;
-    use std::net::SocketAddr;
+    use std::{net::SocketAddr, pin::Pin, task};
+    use tokio_stream::wrappers::IntervalStream;
 
     pub(super) type Metrics = http::balance::MetricFamilies<Labels>;
 
-    pub fn layer<B, R: Clone, N>(
+    pub(super) type Resolve = recover::Resolve<ResolveRecover, DnsResolve>;
+
+    pub fn layer<B, N>(
         metrics: Metrics,
         dns: dns::Resolver,
-        recover: R,
-    ) -> impl svc::Layer<
-        N,
-        Service = http::NewBalance<B, Params, recover::Resolve<R, DnsResolve>, NewIntoTarget<N>>,
-    > {
-        let resolve = recover::Resolve::new(recover, DnsResolve::new(dns));
+        backoff: ExponentialBackoff,
+    ) -> impl svc::Layer<N, Service = http::NewBalance<B, Params, Resolve, NewIntoTarget<N>>> {
+        let resolve = recover::Resolve::new(ResolveRecover::new(backoff), DnsResolve::new(dns));
+
         svc::layer::mk(move |inner| {
             http::NewBalance::new(
                 NewIntoTarget { inner },
@@ -293,6 +279,98 @@ mod balance {
         addr: String,
     }
 
+    /// A [`Recover`] implementation that respects DNS resolution errors.
+    ///
+    /// When a DNS resolution fails, this will log the error and use the TTL, if there is one,
+    /// to drive re-resolution attempts. It defaults to an exponential backoff with jitter for
+    /// other errors.
+    #[derive(Clone)]
+    pub(super) struct ResolveRecover {
+        backoff: ExponentialBackoff,
+    }
+
+    /// A [`Stream`] used for control-plane client's error recovery.
+    #[pin_project::pin_project(project = ResolveBackoffProj)]
+    pub(super) enum ResolveBackoff {
+        /// A DNS-resolution error occurred.
+        ///
+        /// This will backoff at a regular interval according to the negative TTL.
+        NegativeTtl(#[pin] IntervalStream),
+        /// A jittered exponential backoff.
+        ExponentialBackoff(#[pin] ExponentialBackoffStream),
+    }
+
+    // === impl ResolveRecover ===
+
+    impl ResolveRecover {
+        /// Returns a new [`ResolveRecover`].
+        pub fn new(config: ExponentialBackoff) -> Self {
+            Self { backoff: config }
+        }
+    }
+
+    impl Recover for ResolveRecover {
+        type Backoff = ResolveBackoff;
+        fn recover(
+            &self,
+            error: linkerd_error::Error,
+        ) -> Result<Self::Backoff, linkerd_error::Error> {
+            let ResolveRecover { backoff } = self;
+
+            tracing::warn!(error, "Failed to resolve control-plane component");
+
+            // If we are recovering due to a DNS resolution error, check for a negative TTL.
+            if let Some(e) = crate::errors::cause_ref::<dns::ResolveError>(&*error) {
+                if let Some(ttl) = e.negative_ttl() {
+                    let mut interval = tokio::time::interval(ttl);
+
+                    // `tokio::time::interval()`'s documentation states that the first tick
+                    // completes immediately. To respect the prescribed negative-ttl and avoid
+                    // immediately retrying, we reset the interval before returning it.
+                    interval.reset();
+
+                    let stream = IntervalStream::new(interval);
+                    return Ok(ResolveBackoff::NegativeTtl(stream));
+                }
+            }
+
+            // If the error didn't give us a TTL, use the default jittered backoff.
+            Ok(ResolveBackoff::ExponentialBackoff(backoff.stream()))
+        }
+    }
+
+    // === impl ResolveBackoff ===
+
+    impl Stream for ResolveBackoff {
+        /// No items are yielded in this stream.
+        type Item = ();
+
+        fn poll_next(
+            self: Pin<&mut Self>,
+            cx: &mut task::Context<'_>,
+        ) -> task::Poll<Option<Self::Item>> {
+            use ResolveBackoffProj::{ExponentialBackoff, NegativeTtl};
+
+            match self.project() {
+                NegativeTtl(stream) => {
+                    // We can discard the `Instant`s that IntervalStream yields.
+                    let discard = |opt: Option<_>| opt.map(drop);
+                    stream.poll_next(cx).map(discard)
+                }
+                ExponentialBackoff(stream) => stream.poll_next(cx),
+            }
+        }
+
+        fn size_hint(&self) -> (usize, Option<usize>) {
+            use ResolveBackoff::{ExponentialBackoff, NegativeTtl};
+
+            match self {
+                NegativeTtl(stream) => stream.size_hint(),
+                ExponentialBackoff(stream) => stream.size_hint(),
+            }
+        }
+    }
+
     // === impl NewIntoTarget ===
 
     impl<N: svc::NewService<ControlAddr>> svc::NewService<ControlAddr> for NewIntoTarget<N> {