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

## 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`:

```rust
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:

```rust
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>` field is only dropped
after that body returns:

```rust
pub struct OwnedMutexGuard {
...
lock: Arc>,
}

impl Drop for OwnedMutexGuard {
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:

```rust
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> = 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>>` 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.

Start with TransportConnector::release_stream, reused_stream, and ConnectionPool::{get, put, idle_poll}; trace the OwnedMutexGuard lifetime around Arc::try_unwrap. Run connectors::tests::test_connect_uds and the fresh-runtime reproductions described in the issue. Done means healthy pooled connections are reliably reused without the race or intermittent assertion failure.

Written by the indexing model from the issue text.