Fix reused_stream losing healthy pooled connections (race with idle poller)

[sinhaparth5]

Fixes #998.

Root cause

TransportConnector::reused_stream waits for the idle poller's OwnedMutexGuard to be released (by briefly locking the pooled stream's mutex), then calls Arc::try_unwrap, assuming the strong count is back down to 1.

That assumption races with tokio::sync::OwnedMutexGuard::drop, which releases the semaphore permit before it drops its own Arc<Mutex<T>> clone (the permit release happens inside the Drop impl's body, and the guard's Arc field is only dropped after that body returns). So there's a window where the waiter in reused_stream gets woken and scheduled on another thread before the poller's task has actually finished dropping its Arc clone. If try_unwrap runs in that window, it returns Err, the healthy pooled connection gets thrown away, and the caller dials a new connection instead of reusing it.

Nothing is corrupted or leaked. The only symptom is silently lost keepalive reuse, which on a busy proxy means extra TCP/TLS handshakes on the upstream path under load.

Fix

Stop depending on the Arc strong count. The pooled stream is now wrapped as Arc<Mutex<ReusableStream>>, where ReusableStream(Option<Stream>), and reused_stream holds the lock guard and take()s the stream out of the Option instead of calling Arc::try_unwrap. Holding the mutex guard is enough on its own for exclusivity, regardless of whether the poller's leftover Arc clone has finished dropping yet.

Testing

• cargo test -p pingora-core --features rustls passes.
• Reproduced the race on an isolated 8-vCPU GCE VM (torn down afterward): looping connectors::tests::test_connect_uds as 400 fresh processes (fresh multi-threaded runtime per iteration, 8-way parallel, matching the repro methodology in reused_stream can miss a pooled connection when try_unwrap races the idle poller #998) gave 11/400 (2.75%) failures on the parent commit and 0/400 on this fix, all at the same assert!(reused) site described in the issue.