fix(breaker): prevent latent busy-spin in closed-state drain loop (#4481)

* fix(breaker): prevent latent busy-spin in closed-state drain loop

A latent busy-spin exists in the closed-state backoff drain loop:
`_ = self.rsps.recv() => continue` discards the recv() result, so when
the response channel closes recv() returns None immediately on every
iteration and the loop spins, saturating a CPU core.

Currently this is unreachable because gate::Rx and mpsc::Sender live in
the same Gate<BroadcastClassification<...>> service struct and drop
together in a single synchronous destructor chain, so gate.lost() always
fires when the channel is closing. However, gate::Rx is a
watch::Receiver, implementing Clone. If a future change cloned it for a
legitimate purpose (monitoring gate state, exposing it to an admin
endpoint, observability, etc), the clone would keep gate.lost() from
resolving after the endpoint service is dropped, while the mpsc channel
would close normally once in-flight requests drain. The breaker task
would then busy-spin instead of terminating. [NB: the reason for this
is that a clone would extend the lifetime of the receiver, and for
gate.lost() to fire the internal Sender::closed() requires that all
receivers are gone]

Check the recv() result and propagate None as Err(()) to terminate the
task cleanly, mirroring the pattern already used in open() and
probation(). This makes the breaker self-contained, terminating on its
own channel state rather than depending on the fragility of the
ownership structure, where currently gate::Rx and mpsc::Sender happen to
share a destructor chain.

Signed-off-by: Alejandro Martinez Ruiz <amr@buoyant.io>

* test(breaker): regression test for channel close with surviving gate::Rx

Verify that the breaker task terminates when the response channel closes
while a clone of gate::Rx outlives the endpoint service.

We build a minimal Gate, clone gate::Rx, trip the breaker via status 500
into closed state then let the service drop. The surviving gate::Rx
clone should prevent gate.lost() from firing, so the breaker must
terminate via the recv() call, returning None in the closed state drain
loop.

Without the channel close check in closed() this test would hang
(spinning on recv() returning None indefinitely).

Signed-off-by: Alejandro Martinez Ruiz <amr@buoyant.io>

---------

Signed-off-by: Alejandro Martinez Ruiz <amr@buoyant.io>

Author: unleashed

linkerd/app/outbound/src/http/breaker/consecutive_failures.rs

@@ -76,8 +76,11 @@ impl ConsecutiveFailures {
             loop {
                 tokio::select! {
                     _ = backoff.next() => break,
-                    // Ignore responses while the breaker is shut.
-                    _ = self.rsps.recv() => continue,
+                    // Ignore responses while the breaker is shut, but
+                    // terminate if the channel is closed. recv() on a
+                    // closed channel returns None immediately, which would
+                    // otherwise busy-spin this loop.
+                    rsp = self.rsps.recv() => { rsp.ok_or(())?; },
                     _ = self.gate.lost() => return Err(()),
                 }
             }
@@ -105,7 +108,12 @@ impl ConsecutiveFailures {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use tokio::time;
+    use linkerd_app_core::{
+        proxy::http::{self, classify::BroadcastClassification, BoxBody},
+        svc::{self, ServiceExt},
+        Error,
+    };
+    use tokio::{task::yield_now, time};
     use tokio_test::{assert_pending, task};
 
     #[tokio::test(flavor = "current_thread", start_paused = true)]
@@ -208,4 +216,113 @@ mod tests {
         assert_pending!(task.poll());
         assert!(params.gate.is_open());
     }
+
+    #[tokio::test(flavor = "current_thread", start_paused = true)]
+    async fn closed_state_terminates_on_channel_close() {
+        let _trace = linkerd_tracing::test::trace_init();
+
+        let (params, gate, rsps) = gate::Params::channel(1);
+        let send = |res: Result<http::StatusCode, http::StatusCode>| {
+            params
+                .responses
+                .try_send(classify::Class::Http(res))
+                .unwrap()
+        };
+
+        let backoff = ExponentialBackoff::try_new(
+            time::Duration::from_secs(1),
+            time::Duration::from_secs(100),
+            0.0,
+        )
+        .expect("backoff params are valid");
+
+        // max_failures=1: a single failure trips the breaker into closed state.
+        let breaker = ConsecutiveFailures::new(1, backoff, gate, rsps);
+        let mut task = task::spawn(breaker.run());
+
+        assert_pending!(task.poll());
+        assert!(params.gate.is_open());
+
+        // Trip the breaker into the closed state.
+        send(Err(http::StatusCode::INTERNAL_SERVER_ERROR));
+        assert_pending!(task.poll());
+        assert!(params.gate.is_shut(), "should be in closed state");
+
+        // Drop the response sender while in the closed state drain loop.
+        // The breaker must terminate rather than busy-spinning on None from
+        // the closed channel.
+        drop(params.responses);
+        assert!(task.poll().is_ready());
+    }
+
+    /// This test uses `tokio::spawn` (real task) instead of
+    /// `tokio_test::task::spawn` because the Gate + BroadcastClassification
+    /// service stack requires an actual task context for `oneshot()` to drive
+    /// response classification through the mpsc channel.
+    #[tokio::test(flavor = "current_thread", start_paused = true)]
+    async fn closed_state_terminates_when_service_drops_but_gate_rx_survives() {
+        let _trace = linkerd_tracing::test::trace_init();
+
+        let (params, gate, rsps) = gate::Params::channel(1);
+        // Keep an extra gate receiver alive to model a future stack change that
+        // retains gate state after the endpoint service itself is dropped.
+        let leaked_gate = params.gate.clone();
+
+        let backoff = ExponentialBackoff::try_new(
+            time::Duration::from_secs(1),
+            time::Duration::from_secs(100),
+            0.0,
+        )
+        .expect("backoff params are valid");
+
+        let breaker = ConsecutiveFailures::new(1, backoff, gate, rsps);
+        let breaker = tokio::spawn(breaker.run());
+
+        let svc = svc::Gate::new(
+            params.gate,
+            BroadcastClassification::<classify::Response, _>::new(
+                params.responses,
+                svc::mk(|_: http::Request<BoxBody>| async move {
+                    Ok::<_, Error>(
+                        http::Response::builder()
+                            .status(http::StatusCode::INTERNAL_SERVER_ERROR)
+                            .body(BoxBody::default())
+                            .unwrap(),
+                    )
+                }),
+            ),
+        );
+
+        let req = http::Request::builder()
+            .extension(classify::Response::default())
+            .body(BoxBody::default())
+            .unwrap();
+        let rsp = svc.oneshot(req).await.expect("service must succeed");
+        drop(rsp);
+
+        for _ in 0..16 {
+            if leaked_gate.is_shut() {
+                break;
+            }
+            yield_now().await;
+        }
+        assert!(
+            leaked_gate.is_shut(),
+            "breaker should enter the closed state"
+        );
+
+        for _ in 0..16 {
+            if breaker.is_finished() {
+                break;
+            }
+            yield_now().await;
+        }
+
+        assert!(
+            breaker.is_finished(),
+            "breaker should terminate when the response channel closes, \
+             even if a gate::Rx clone survives"
+        );
+        breaker.await.expect("breaker task must not panic");
+    }
 }