buffer's worker closed unexpectedly 与 buffered service failed

tower = { version = "=0.5.3", features = ["buffer", "util"] }
tokio = { version = "=1.54.1", features = ["full"] }

// tests/shared_buffer.rs
use std::future::Ready;
use std::sync::OnceLock;
use tower::buffer::Buffer;
use tower::{BoxError, Service, ServiceExt, service_fn};

type Client = Buffer<u32, Ready<Result<u32, BoxError>>>;

fn echo(x: u32) -> Ready<Result<u32, BoxError>> {
  std::future::ready(Ok(x))
}

// One client for the whole test binary, created by whichever test gets here first.
fn client() -> Client {
  static CLIENT: OnceLock<Client> = OnceLock::new();
  CLIENT.get_or_init(|| Buffer::new(service_fn(echo), 16)).clone()
}

async fn roundtrip(mut c: Client, x: u32) {
  match c.ready().await {
    Ok(svc) => assert_eq!(svc.call(x).await.unwrap(), x),
    Err(e) => panic!("ready failed: {e} ({e:?})"),
  }
}

#[tokio::test]
async fn first() {
  roundtrip(client(), 1).await;
}

#[tokio::test]
async fn second() {
  roundtrip(client(), 2).await;
}

running 2 tests
. 1/2
second --- FAILED

failures:

---- second stdout ----

thread 'second' (142463) panicked at tests/shared_buffer.rs:21:15:
ready failed: buffer's worker closed unexpectedly (Closed)
note: run with `RUST_BACKTRACE=1` environment variable to display a backtrace


failures:
    second

test result: FAILED. 1 passed; 1 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.00s

// tests/shared_buffer_fixed.rs：只有 client() 不同
use tokio::runtime::Runtime;

// The worker runs on a runtime that lives as long as the test binary.
fn client() -> Client {
  static RT: OnceLock<Runtime> = OnceLock::new();
  static CLIENT: OnceLock<Client> = OnceLock::new();
  CLIENT
    .get_or_init(|| {
      let rt = RT.get_or_init(|| {
        tokio::runtime::Builder::new_multi_thread()
          .worker_threads(1)
          .build()
          .unwrap()
      });
      let (client, worker) = Buffer::pair(service_fn(echo), 16);
      rt.spawn(worker);
      client
    })
    .clone()
}

tower = { version = "=0.5.3", features = ["buffer", "reconnect", "util"] }
tokio = { version = "=1.54.1", features = ["full"] }

use std::future::{Ready, ready};
use std::sync::Arc;
use std::sync::atomic::{AtomicU32, Ordering};
use std::task::{Context, Poll};
use tower::buffer::Buffer;
use tower::reconnect::Reconnect;
use tower::{BoxError, Service, ServiceExt, service_fn};

// A connection. Connection #1 breaks after serving one request:
// its poll_ready reports the broken connection as an error.
struct Conn {
  id: u32,
  served: u32,
}

impl Service<u32> for Conn {
  type Response = String;
  type Error = BoxError;
  type Future = Ready<Result<String, BoxError>>;

  fn poll_ready(&mut self, _: &mut Context<'_>) -> Poll<Result<(), BoxError>> {
    if self.id == 1 && self.served >= 1 {
      return Poll::Ready(Err("connection reset".into()));
    }
    Poll::Ready(Ok(()))
  }

  fn call(&mut self, req: u32) -> Self::Future {
    self.served += 1;
    ready(Ok(format!("req {req} served by conn #{}", self.id)))
  }
}

async fn send<S>(svc: &mut S, req: u32)
where
  S: Service<u32, Response = String, Error = BoxError>,
{
  match svc.ready().await {
    Ok(s) => match s.call(req).await {
      Ok(rsp) => println!("  {rsp}"),
      Err(e) => println!("  req {req}: call failed: {e}"),
    },
    Err(e) => println!("  req {req}: ready failed: {e}"),
  }
}

#[tokio::main(flavor = "current_thread")]
async fn main() {
  println!("Buffer over a connection:");
  let mut svc = Buffer::new(Conn { id: 1, served: 0 }, 8);
  for req in 1..=3 {
    send(&mut svc, req).await;
  }

  println!("Buffer over Reconnect:");
  let next_id = Arc::new(AtomicU32::new(1));
  let mk = service_fn(move |_: ()| {
    let id = next_id.fetch_add(1, Ordering::SeqCst);
    ready(Ok::<_, BoxError>(Conn { id, served: 0 }))
  });
  let mut svc = Buffer::new(Reconnect::new(mk, ()), 8);
  for req in 1..=3 {
    send(&mut svc, req).await;
  }
}

Buffer over a connection:
  req 1 served by conn #1
  req 2: call failed: buffered service failed: connection reset
  req 3: ready failed: buffered service failed: connection reset
Buffer over Reconnect:
  req 1 served by conn #1
  req 2 served by conn #2
  req 3 served by conn #2