Baike.dev
All toolsAI codingTrendingOpen sourceNewsSubmit
Log in
Back to tool/Back to issues
#998·pingora

reused_stream can miss a pooled connection when try_unwrap races the idle poller

Author: ShanireZCreated Sep 8, 2026Updated Sep 8, 2026

Describe the bug

TransportConnector::reused_stream waits for the idle poller to let go of a pooled connection by taking the mutex, and then takes ownership of the stream with Arc::try_unwrap:

Some(s) => {
    debug!("find reusable stream, trying to acquire it");
    {
        let _ = s.lock().await;
    } // wait for the idle poll to release it
    match Arc::try_unwrap(s) {

Holding the mutex is not enough to know the strong count is back to 1. release_stream hands a second Arc to the spawned poller:

let stream = Arc::new(Mutex::new(stream));
let locked_stream = stream.clone().try_lock_owned().unwrap(); // safe as we just created it
let (notify_close, watch_use) = self.connection_pool.put(&meta, stream);
...
rt.spawn(async move {
    pool.idle_poll(locked_stream, &meta, idle_timeout, notify_close, watch_use)
        .await;
});

That second Arc lives inside the OwnedMutexGuard. In tokio the guard releases the semaphore in the body of Drop::drop, and its lock: Arc<Mutex<T>> field is only dropped after that body returns:

pub struct OwnedMutexGuard<T: ?Sized> {
    ...
    lock: Arc<Mutex<T>>,
}

impl<T: ?Sized> Drop for OwnedMutexGuard<T> {
    fn drop(&mut self) {
        self.lock.s.release(1);
        ...
    }
}

So there is a window where the waiter has already been woken by release(1) while the poller task has not yet dropped its Arc. If the waiter reaches Arc::try_unwrap inside that window it gets Err, reused_stream logs failed to acquire reusable stream and returns None, and the caller dials a new connection even though the pooled one was perfectly healthy.

Nothing is leaked or corrupted, so the only symptom is keepalive silently not happening. I ran into it as an intermittent CI failure of connectors::tests::test_connect_uds (assert!(reused)), which is how I started digging.

Pingora info

Pingora version: main @ 09696b5, and 0.8.1 — release_stream, reused_stream and ConnectionPool::{get, put, idle_poll} are byte-identical between the two Rust version: cargo 1.98.0 (797e8a9bc 2026-08-05) Operating system version: Debian 13 (trixie), x86_64

Steps to reproduce

Two reproductions below. The first goes through TransportConnector; the second has no pingora in it at all and shows the pattern on its own.

One thing worth knowing before you try: what decides whether you see this at all is a fresh multi-threaded runtime per iteration, which is what #[tokio::test(flavor = "multi_thread")] gives you. Looping inside one long-lived runtime almost never hits it. Same total iteration count in both cases, second reproduction below:

fresh runtime each iteration one shared runtime
--cpus=2 1.92% 0.00% (0 of 12800)
24 CPUs 8.45% 0.02% (3 of 12800)

I do not have an explanation for the size of that gap, only the measurement. It cost me a while: my first attempt at the connector reproduction below looped inside a single runtime and came back clean after 18,400 iterations spread over 0.5, 1, 2 and 24 CPUs, which I nearly read as the theory being wrong.

1. Through the connector

A small binary with a path dependency on pingora-core, plus a log::Log implementation that counts the messages of the four paths through reused_stream that return None (find reusable stream, failed to acquire reusable stream, the test_reusable_stream messages, and the matches_fd messages). Each iteration builds a fresh multi-threaded runtime, connects to a mock UDS server that writes it works! and holds the connection open, does read_exact, then release_stream followed immediately by get_stream.

With 48 threads x 100 iterations in a container limited to --cpus=2:

reused == true                     4552
reused == false                     248
  failed to acquire reusable stream 248
  test_reusable_stream (3 branches)   0
  matches_fd mismatch                 0
find reusable stream               4800   (i.e. the connection was pooled every time)
matches_fd ok                      4552

4552 + 248 = 4800, so every attempt is accounted for. Every single reuse failure came from Arc::try_unwrap; the other paths never fired.

Failure rate scales with oversubscription, all at --cpus=2: 4 threads 0.75%, 16 threads 1.50%, 48 threads 4.69%.

2. Without pingora

Arc + tokio::sync::Mutex + oneshot, in the same shape as release/reuse:

use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::{oneshot, Mutex};

static OK: AtomicUsize = AtomicUsize::new(0);
static FAIL: AtomicUsize = AtomicUsize::new(0);

async fn one_round() {
    let s: Arc<Mutex<u8>> = Arc::new(Mutex::new(7));

    // the second Arc, living inside the guard
    let guard = s.clone().lock_owned().await;

    let (tx, rx) = oneshot::channel::<()>();
    tokio::spawn(async move {
        let _ = rx.await;
        drop(guard); // releases the lock first, drops the Arc after
    });

    // tell it to finish, then wait for the lock, exactly like reused_stream does
    let _ = tx.send(());
    {
        let _ = s.lock().await;
    }

    match Arc::try_unwrap(s) {
        Ok(_) => OK.fetch_add(1, Ordering::Relaxed),
        Err(_) => FAIL.fetch_add(1, Ordering::Relaxed),
    };
}

fn main() {
    let mut ths = Vec::new();
    for _ in 0..64 {
        ths.push(std::thread::spawn(|| {
            for _ in 0..200 {
                let rt = tokio::runtime::Builder::new_multi_thread()
                    .enable_all()
                    .build()
                    .unwrap();
                rt.block_on(one_round());
            }
        }));
    }
    for t in ths {
        let _ = t.join();
    }
    println!("ok {} fail {}", OK.load(Ordering::Relaxed), FAIL.load(Ordering::Relaxed));
}

tokio 1.53.1, exactly the code above (64 threads x 200 iterations = 12800), varying how many CPUs the container gets:

--cpus=1    ok 12744   fail 56     0.44%
--cpus=2    ok 12554   fail 246    1.92%
--cpus=4    ok 12205   fail 595    4.65%
24 CPUs     ok 11719   fail 1081   8.45%

More real parallelism makes it easier to hit, which is what you would expect if the waiter needs to get to try_unwrap while the other task is still inside the guard's drop.

Expected results

After release_stream puts a healthy connection in the pool, the next get_stream for the same peer reuses it.

Observed results

Some of the time it does not, and a new connection is dialled instead. In my measurements every one of those cases was Arc::try_unwrap returning Err while the connection itself was fine.

Additional context

#201 asked whether error! is the right level for failed to acquire reusable stream, on the assumption that it fires when the peer drops an idle connection. That is not what this branch catches. A peer that closed the connection is caught further down by test_reusable_stream, which returns None without logging an error, and in the runs above that branch never fired once while try_unwrap failed 248 times. So the error level looks right to me, and lowering it would hide this instead of fixing it.

This is also a second, independent reason for connectors::tests::test_connect_uds to fail, separate from the short read in #967. The assertion that trips here is assert!(reused) rather than the byte comparison, and the two have different fixes.

As for a fix, the part that looks fragile to me is depending on the strong count at all. Pooling something like Arc<Mutex<Option<Stream>>> would let reused_stream take() the stream out from under the guard it already holds, and the poller's Arc would stop mattering. Having the poller drop its guard before signalling that it is done would also close the window, but that seems easier to get subtly wrong. Happy to put a PR together if you have a preference on which direction you want.

Source: cloudflare/pingora

View original on GitHubView discussion on GitHub