RTCache lookup cancellation can leave same-key requests blocked on a stale coalescing lock

## Describe the bug

`pingora-memory-cache::RTCache::get()` can leak a per-key lookup-coalescing lock if the cache-miss writer future is cancelled while it is awaiting the user-provided `Lookup::lookup()` callback.

The vulnerable state transition is:

```text
cache miss
-> insert zero-permit CacheLock into RTCache.lockers for this key
-> await user async callback CB::lookup(key, extra)
-> caller cancels/drops the RTCache::get() future before lookup returns
-> cleanup after the await is skipped
-> later same-key requests observe the stale lock as an active writer
```

With `lock_age = None` and `lock_timeout = None`, later same-key callers wait forever on the leaked zero-permit semaphore. With `lock_timeout = Some(...)`, callers can eventually escape by doing a direct lookup, but the stale lock is not repaired and the timeout path returns without populating the cache. With `lock_age = Some(...)`, aged locks may be bypassed, but the stale map entry can still remain and later lookups may continue to lose normal cache/coalescing behavior.

This is a cancellation-correctness issue because shared in-progress state is published before an arbitrary external async callback, while the only cleanup and waiter wakeup path remains after that cancellation point.

### Root cause

The root cause is visible in `pingora-memory-cache/src/read_through.rs`. All source links below are pinned to commit `e6e677fe9b58555140ab7bd14feff035392b3530`.

1. `RTCache::get()` is a normal caller-owned async future. A caller can drop it through a timeout, a losing `select!` branch, task abort, parent future cancellation, or runtime shutdown:

[`pingora-memory-cache/src/read_through.rs#L150-L155`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L150-L155)

```rust
pub async fn get(
&self,
key: &K,
ttl: Option,
extra: Option<&S>,
) -> (Result>, CacheStatus) {
```

2. On a miss with no existing lock, `RTCache::get()` inserts a new per-key `CacheLock` into the shared `lockers` map before it awaits the user callback:

[`pingora-memory-cache/src/read_through.rs#L193-L198`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L193-L198)

```rust
None => {
let new_lock = CacheLock::new_arc();
let new_lock2 = new_lock.clone();
lockers.insert(hashed_key, new_lock2);
(Some(new_lock), None)
}
```

The inserted lock starts with zero permits:

[`pingora-memory-cache/src/read_through.rs#L36-L42`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L36-L42)

```rust
pub fn new_arc() -> Arc {
Arc::new(CacheLock {
lock: Semaphore::new(0),
lock_start: Instant::now(),
})
}
```

3. Readers that find the lock treat it as an active writer and wait for a permit. If no `lock_timeout` is configured, this await has no completion source once the writer has been cancelled:

[`pingora-memory-cache/src/read_through.rs#L203-L228`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L203-L228)

```rust
if let Some(my_lock) = my_read {
/* another task will do the lookup */

/* if available_permits > 0, writer is done */
if my_lock.lock.available_permits() == 0 {
/* block here to wait for writer to finish lookup */
let lock_fut = my_lock.lock.acquire();
let timed_out = match self.lock_timeout {
Some(t) => pingora_timeout::timeout(t, lock_fut).await.is_err(),
None => {
let _ = lock_fut.await;
false
}
};
```

4. The writer crosses the cancellation point while the shared lock is already published:

[`pingora-memory-cache/src/read_through.rs#L252-L255`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L252-L255)

```rust
} else {
/* this one will do the look up, either because it gets the write lock or the read
* lock age is reached */
let value = CB::lookup(key, extra).await;
```

`Lookup::lookup()` is implemented by the caller and can be arbitrary async work, including DNS, database, or backend I/O:

[`pingora-memory-cache/src/read_through.rs#L75-L80`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L75-L80)

```rust
pub trait Lookup {
/// Return a value and an optional TTL for the given key.
async fn lookup(
key: &K,
extra: Option<&S>,
) -> Result<(T, Option), Box>
```

5. The only normal release and removal path runs after that await. If the future is dropped while `CB::lookup()` is pending, this code is skipped:

[`pingora-memory-cache/src/read_through.rs#L270-L279`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L270-L279)

```rust
if let Some(my_write) = my_write {
/* add permit so that reader can start. Any number of permits will do,
* since readers will return permits right away. */
my_write.lock.add_permits(10);

{
// remove the lock from locker
let mut lockers = self.lockers.write();
lockers.remove(&hashed_key);
} // write lock dropped here
}
```

This creates the leaked state:

```text
lockers[hashed_key] still present
Semaphore permits still 0
no cache value inserted
no live writer future remains
no live future remains that will call add_permits(10)
no live future remains that will remove lockers[hashed_key]
```

6. Timeout and lock-age settings are partial escape hatches, not full repairs. If `lock_timeout` fires, the caller performs a direct lookup and returns without inserting into the cache:

[`pingora-memory-cache/src/read_through.rs#L217-L226`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L217-L226)

```rust
if timed_out {
let value = CB::lookup(key, extra).await;
return match value {
Ok((v, _ttl)) => (Ok(v), cache_state),
Err(e) => {
let mut err = Error::new_str(LOOKUP_ERR_MSG);
err.set_cause(e);
(Err(err), cache_state)
}
};
}
```

If `lock_age` treats the stale lock as too old, `my_write` remains `None`, and successful lookup results are not inserted because cache insertion is gated on `my_write.is_some()`:

[`pingora-memory-cache/src/read_through.rs#L174-L180`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L174-L180)

```rust
Some(lock) => {
/* There is an ongoing lookup to the same key */
if lock.too_old(self.lock_age.as_ref()) {
(None, None)
} else {
(None, Some(lock))
}
}
```

[`pingora-memory-cache/src/read_through.rs#L257-L261`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-memory-cache/src/read_through.rs#L257-L261)

```rust
Ok((v, new_ttl)) => {
/* Don't put() if lock ago too old, to avoid too many concurrent writes */
if my_write.is_some() {
self.inner.force_put(key, v.clone(), new_ttl.or(ttl));
}
```

## Pingora info

Please include the following information about your environment:

**Pingora version**: audited and reproduced against commit `e6e677fe9b58555140ab7bd14feff035392b3530`
**Rust version**: not captured in the shared reproduction log
**Operating system version**: reproduced as a Rust unit test on Linux x86 (`server36`)

## Steps to reproduce

The following is a focused whitebox reproduction. It intentionally asserts the current bad state, so the test passing means the cancellation window is reachable in the current implementation. After a fix, the core assertions should be inverted into a regression property such as: a cancelled writer removes or invalidates the coalescing lock, wakes waiters, and lets a later same-key request become a new writer or otherwise make bounded progress.

Steps:

1. Apply the test-only code below inside the existing `#[cfg(test)] mod tests` in `pingora-memory-cache/src/read_through.rs`.
2. Run the targeted test command shown below.
3. Observe that the cancelled writer leaves a stale zero-permit lock in `RTCache.lockers`, and the next same-key `get()` remains blocked on that leaked lock.

Full whitebox test code for pingora-memory-cache/src/read_through.rs

```rust
#[derive(Clone, Debug)]
struct CancelledLookupOpt {
entered: Arc,
used: Arc,
}

struct CancelledLookupCB();

#[async_trait]
impl Lookup for CancelledLookupCB {
async fn lookup(
_key: &i32,
extra: Option<&CancelledLookupOpt>,
) -> Result<(i32, Option), Box> {
let extra = extra.expect("test lookup must receive coordination state");
let used = extra.used.fetch_add(1, atomic::Ordering::SeqCst) + 1;
if used == 1 {
extra.entered.notify_one();
std::future::pending::<()>().await;
unreachable!("first lookup is cancelled while pending");
}
Ok((used, None))
}
}

#[tokio::test]
async fn test_cancelled_lookup_keeps_coalescing_lock_blocking_next_get() {
let cache: Arc> =
Arc::new(RTCache::new(10, None, None));
let opt = CancelledLookupOpt {
entered: Arc::new(tokio::sync::Notify::new()),
used: Arc::new(AtomicI32::new(0)),
};
assert!(cache.lockers.read().is_empty());

let writer_cache = cache.clone();
let writer_opt = opt.clone();
let writer = tokio::spawn(async move {
writer_cache.get(&1, None, Some(&writer_opt)).await
});

opt.entered.notified().await;
assert_eq!(
opt.used.load(atomic::Ordering::SeqCst),
1,
"the first cache miss writer must enter lookup before cancellation"
);

writer.abort();
assert!(
writer.await.unwrap_err().is_cancelled(),
"aborting the writer task is the cancellation source"
);

let hashed_key = cache.inner.hasher.hash_one(&1);
let leaked_lock = cache
.lockers
.read()
.get(&hashed_key)
.cloned()
.expect("cancelled writer leaves the coalescing lock installed");
assert_eq!(
leaked_lock.lock.available_permits(),
0,
"cancelled writer never reaches add_permits(), so waiters cannot be released"
);

let next_get = tokio::time::timeout(
Duration::from_millis(50),
cache.get(&1, None, Some(&opt)),
)
.await;

assert!(
next_get.is_err(),
"the next same-key get is still waiting on the leaked coalescing lock"
);
assert_eq!(
opt.used.load(atomic::Ordering::SeqCst),
1,
"the second get never becomes a replacement writer, so lookup is not called again"
);
}
```

Targeted command:

```bash
cargo test -p pingora-memory-cache test_cancelled_lookup_keeps_coalescing_lock_blocking_next_get -- --nocapture
```

Observed result:

```text
Finished `test` profile [unoptimized + debuginfo] target(s) in 14.48s
Running unittests src/lib.rs (target/debug/deps/pingora_memory_cache-32c430ff3d060b71)

running 1 test
test read_through::tests::test_cancelled_lookup_keeps_coalescing_lock_blocking_next_get ... ok

test result: ok. 1 passed; 0 failed; 0 ignored; 0 measured; 20 filtered out; finished in 0.05s
```

Why this reproduces the issue:

- The test starts from a healthy `RTCache` with an empty `lockers` map.
- The first same-key miss becomes the writer and enters `CB::lookup()`.
- The test callback uses `Notify` to deterministically prove the writer is suspended inside the lookup callback, which is the cancellation window after lock insertion and before cleanup.
- `writer.abort()` is the real cancellation source. It drops the in-flight `RTCache::get()` future while that future owns the only remaining cleanup path.
- The test then reads `cache.lockers` directly and proves the bad shared state: the per-key lock is still installed and has `0` available permits.
- The next same-key `get()` does not become a replacement writer. It waits on the stale zero-permit lock and is observed with a short timeout.
- The callback counter remains `1`, proving that the second request never reached `CB::lookup()` and therefore could not repopulate or repair the key.

This timeout is only used as a deadlock guard around the final observation. The bad state itself is proven by direct whitebox assertions on `RTCache.lockers` and the leaked semaphore permits.

## Expected results

Cancelling a cache-miss writer should not leave a shared "lookup in progress" marker that no live future can complete.

More specifically, after a writer is cancelled while populating a key, later same-key callers should still have a bounded-progress path. They should either:

- observe a cached value from a completed writer;
- observe a completed error/failure state;
- become a new writer and retry the lookup;
- or wait on a lock that is guaranteed to be released by a live owner.

The cache should not keep a stale zero-permit `CacheLock` in `RTCache.lockers` after the future responsible for releasing and removing that lock has been cancelled.

## Observed results

The whitebox test observes this bad state after cancelling the cache-miss writer:

```text
RTCache.lockers[hashed_key] exists
leaked_lock.lock.available_permits() == 0
second same-key RTCache::get() does not call CB::lookup()
second same-key RTCache::get() remains blocked on the leaked lock
```

For `RTCache::new(size, None, None)`, this is a per-key liveness failure: future same-key requests can wait indefinitely.

For configurations that use `lock_timeout` or `lock_age`, the impact is different but still problematic. The affected key can repeatedly bypass normal coalescing/cache insertion or remain associated with a stale lock entry, defeating the load-shedding behavior that `RTCache` is intended to provide for async external lookups.

## Additional context

The cancellation-safety invariant I expected is:

```text
A cancellable RTCache writer must not be the only owner of cleanup for a
shared per-key in-progress marker that can block future same-key readers.
```

A robust fix could use a small writer guard that owns the inserted lock entry and removes or marks it failed on drop unless the writer reaches the normal completion path. To avoid a cancelled old writer deleting a newer writer's state, removal should compare the stored `Arc` identity or use a generation token. The same guard should also wake waiters on cancellation or failure so they can retry instead of waiting on a zero-permit lock with no owner.

Start in pingora-memory-cache/src/read_through.rs, especially RTCache::get(), CacheLock, and the existing read_through tests. Run cargo test -p pingora-memory-cache test_cancelled_lookup_keeps_coalescing_lock_blocking_next_get -- --nocapture, then turn the reproduction into a regression test showing that cancellation does not leave same-key requests blocked and that subsequent lookup progress is bounded.

Written by the indexing model from the issue text.