Streaming partial cache reads can treat a cancelled cache writer as clean EOF

## Describe the bug

Pingora's streaming partial cache-write path can let a cache reader observe an unfinished, failed cache admission as a normal end-of-body.

The vulnerable state transition is:

```text
cache miss starts streaming cache admission
-> writer publishes a temporary partial object
-> cache write lock is released so readers can consume the partial object
-> an independent reader consumes the prefix from that temporary object
-> writer request is cancelled or dropped before EOF and before HandleMiss::finish()
-> storage contract says the write failed
-> existing partial reader reports Ok(None)
-> proxy/cache response state treats Ok(None) as clean EOF / HttpTask::Done
```

This can make a cache-served response appear complete after only a prefix of the upstream body. The issue is not that a cache fill failed; failed cache fills are expected. The issue is that an already-created partial reader has no way to distinguish "the associated writer reached a real EOF" from "the associated writer disappeared before finishing".

The concrete in-tree demonstration is in `pingora-cache::memory::MemCache`. I realize `MemCache` is documented as testing-only, so the claim is not that every production deployment necessarily uses this backend. The broader issue is the streaming partial-write state-machine contract: `Storage::support_streaming_partial_write()` tells `HttpCache` to unlock readers before cache admission is complete, but the current `HandleHit::read_body() -> Result>` path gives partial readers only `Ok(None)` for EOF unless each backend invents its own aborted-writer error convention. The proxy layer then treats `Ok(None)` as successful response completion.

### Root cause

The root cause is visible across the streaming partial-write contract, `HttpCache`'s early-unlock path, the in-tree partial reader, and the proxy cache response state machine. Source links below are pinned to commit `e6e677fe9b58555140ab7bd14feff035392b3530`.

1. The storage trait explicitly says dropping a miss handler without `finish()` means the write failed:

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

```rust
pub trait HandleMiss {
/// Write the given body to the storage
async fn write_body(&mut self, data: bytes::Bytes, eof: bool) -> Result<()>;

/// Finish the cache admission
///
/// When `self` is dropped without calling this function, the storage should consider this write
/// failed.
async fn finish(
self: Box, // because self is always used as a trait object
) -> Result;
```

So a writer that is cancelled before `finish()` has not produced a clean EOF. It has produced a failed cache write.

2. `support_streaming_partial_write()` is specifically the signal that partially written data can be read, and that the cache can unlock readers:

[`pingora-cache/src/storage.rs#L93-L98`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-cache/src/storage.rs#L93-L98)

```rust
/// Whether this storage backend supports reading partially written data
///
/// This is to indicate when cache should unlock readers
fn support_streaming_partial_write(&self) -> bool {
false
}
```

3. `HttpCache::set_miss_handler()` acts on that capability by releasing the cache write lock before the body write has reached EOF or `finish()`:

[`pingora-cache/src/lib.rs#L922-L951`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-cache/src/lib.rs#L922-L951)

```rust
if inner_enabled.storage.support_streaming_partial_write() {
// If a reader can access partial write, the cache lock can be released here
// to let readers start reading the body.
if let Some(lock_ctx) = inner_enabled.lock_ctx.as_mut() {
let lock = lock_ctx.lock.take();
if let Some(Locked::Write(permit)) = lock {
lock_ctx.cache_lock.release(key, permit, LockStatus::Done);
}
}
// Downstream read and upstream write can be decoupled
let body_reader = inner_enabled
.storage
.lookup_streaming_write(
key,
inner_enabled
.miss_handler
.as_ref()
.expect("miss handler already set")
.streaming_write_tag(),
&inner_enabled.traces.get_miss_span(),
)
.await?;

if let Some((_meta, body_reader)) = body_reader {
inner_enabled.body_reader = Some(body_reader);
} else {
panic!("unable to get body_reader for {:?}", meta);
}
}
```

This makes the partial body visible before cache admission is complete. That is a valid optimization only if reader terminal states distinguish complete EOF from aborted writer.

4. The in-tree `MemCache` backend opts into streaming partial writes and makes temporary partial objects visible to ordinary lookups:

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

```rust
// always prefer partial read otherwise fresh asset will not be visible on expired asset
// until it is fully updated
// no preference on which partial read we get (if there are multiple writers)
if let Some((_, temp_obj)) = self
.temp
.read()
.get(&hash)
.and_then(|map| map.iter().next())
{
hit_from_temp_obj(temp_obj)
```

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

```rust
fn support_streaming_partial_write(&self) -> bool {
true
}
```

5. `MemCache::PartialHit::read()` treats the writer-side sender being dropped as body completion. The comment already notes the missing distinction:

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

```rust
impl PartialHit {
async fn read(&mut self) -> Option {
loop {
let bytes_written = *self.bytes_written.borrow_and_update();
let bytes_end = match bytes_written {
PartialState::Partial(s) => s,
PartialState::Complete(c) => {
// no more data will arrive
if c == self.bytes_read {
return None;
}
c
}
};
...

// wait for more data
if self.bytes_written.changed().await.is_err() {
// err: sender dropped, body is finished
// FIXME: sender could drop because of an error
return None;
}
}
}
}
```

6. Dropping the unfinished miss handler removes the temporary object. Existing readers still hold the body buffer and receiver, but the writer-side sender disappears:

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

```rust
impl Drop for MemMissHandler {
fn drop(&mut self) {
self.temp
.write()
.get_mut(&self.key)
.and_then(|map| map.remove(&self.temp_id.into()));
}
}
```

In this state, the miss handler did not call `finish()`, so the storage contract says the write failed. The existing partial reader nevertheless receives `None`, which is indistinguishable from clean EOF.

7. `ServeFromCache::next_http_task()` converts `None` from a miss body reader into `HttpTask::Done`:

[`pingora-proxy/src/proxy_cache.rs#L2479-L2495`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-proxy/src/proxy_cache.rs#L2479-L2495)

```rust
Self::CacheBodyMiss(should_seek) => {
if *should_seek {
self.maybe_seek_miss_handler(cache, range)?;
}
// safety: caller of enable_miss() call it only if the async_body_reader exist
loop {
if let Some(b) = cache.miss_body_reader().unwrap().read_body().await? {
return Ok(body_task(b, upgraded));
} else {
// EOF from hit handler for body requested
// if multipart, then seek again
if range.should_cache_seek_again() {
self.maybe_seek_miss_handler(cache, range)?;
} else {
*self = Self::DoneMiss;
return Ok(HttpTask::Done);
```

`HttpTask::Done` is the framework's response-finished signal:

[`pingora-core/src/protocols/http/mod.rs#L35-L49`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-core/src/protocols/http/mod.rs#L35-L49)

```rust
pub enum HttpTask {
/// the response header and the boolean end of response flag
Header(Box, bool),
/// A piece of request or response body and the end of request/response boolean flag.
Body(Option, bool),
/// Request or response body bytes that have been upgraded on H1.1, and EOF bool flag.
UpgradedBody(Option, bool),
/// HTTP response trailer
Trailer(Option>),
/// Signal that the response is already finished
Done,
/// Signal that the reading of the response encountered errors.
Failed(pingora_error::BError),
}
```

For H1 downstream proxy tasks, `Done` sets `end_stream = true`:

[`pingora-core/src/protocols/http/v1/server.rs#L1691-L1693`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-core/src/protocols/http/v1/server.rs#L1691-L1693)

```rust
HttpTask::Trailer(_) | HttpTask::Done => {
end_stream = true;
}
```

For H2, `Done` is also treated as the end of the response:

[`pingora-core/src/protocols/http/v2/server.rs#L523-L538`](https://github.com/cloudflare/pingora/blob/e6e677fe9b58555140ab7bd14feff035392b3530/pingora-core/src/protocols/http/v2/server.rs#L523-L538)

```rust
HttpTask::Trailer(Some(trailers)) => {
self.write_trailers(*trailers)?;
true
}
HttpTask::Trailer(None) => true,
HttpTask::Done => true,
HttpTask::Failed(e) => {
return Err(e);
}
...
if end_stream {
// no-op if finished already
self.finish().map_err(|e| e.into_down())?;
}
```

The resulting invariant gap is:

```text
HandleMiss dropped before finish == failed write
PartialHit reader observes dropped writer == Ok(None)
ServeFromCache maps Ok(None) == HttpTask::Done
```

The same terminal value is used for both successful EOF and aborted partial writer.

## 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 behavior, so the test passing means the state-machine gap is reachable in the current implementation. After a fix, the final assertion should be inverted into a regression property such as: after an unfinished partial writer is dropped, an existing partial reader returns an error or another explicit aborted-writer signal, not `Ok(None)`.

Steps:

1. Apply the test-only code below inside the existing `#[cfg(test)] mod test` in `pingora-cache/src/memory.rs`.
2. Run the targeted test command shown below.
3. Observe that an existing partial reader receives `Ok(None)` after the unfinished writer is dropped, even though the miss handler never reached `finish()`.

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

```rust
#[tokio::test]
async fn test_dropped_partial_write_reader_observes_clean_eof() {
static MEM_CACHE: Lazy = Lazy::new(MemCache::new);
let span = &Span::inactive().handle();

let key = CacheKey::new("", "partial-writer-drop", "reader-clean-eof");
let res = MEM_CACHE.lookup(&key, span).await.unwrap();
assert!(res.is_none());

let cache_meta = gen_meta();
let mut miss_handler = MEM_CACHE
.get_miss_handler(&key, &cache_meta, span)
.await
.unwrap();

miss_handler
.write_body(b"partial-body-prefix"[..].into(), false)
.await
.unwrap();

let (cache_meta2, mut hit_handler) = MEM_CACHE.lookup(&key, span).await.unwrap().unwrap();
assert_eq!(
cache_meta.0.internal.fresh_until,
cache_meta2.0.internal.fresh_until
);

let data = hit_handler.read_body().await.unwrap().unwrap();
assert_eq!("partial-body-prefix", data);

drop(miss_handler);

let res = MEM_CACHE.lookup(&key, span).await.unwrap();
assert!(res.is_none());

let data = hit_handler.read_body().await.unwrap();
assert!(
data.is_none(),
"a dropped unfinished partial writer is currently reported as clean EOF"
);
}
```

Targeted command:

```bash
cargo test -p pingora-cache --lib test_dropped_partial_write_reader_observes_clean_eof -- --nocapture
```

Observed result:

```text
Finished `test` profile [unoptimized + debuginfo] target(s) in 21.63s
Running unittests src/lib.rs (target/debug/deps/pingora_cache-ec49d5488b76df96)

running 1 test
test memory::test::test_dropped_partial_write_reader_observes_clean_eof ... ok

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

Why this reproduces the issue:

- The test starts from a healthy cache state: `lookup()` returns `None` for the selected key.
- It creates a real `MemMissHandler`, writes `partial-body-prefix` with `eof = false`, and does not call `finish()`.
- It creates an independent `HitHandler` through `lookup()`, proving the temporary partial object is visible to readers before admission completes.
- The reader successfully consumes the prefix body.
- The test then drops the miss handler. This models cancellation or request-task drop while the writer owns an unfinished cache admission.
- A fresh `lookup()` for the same key returns `None`, proving the temporary object was removed by `MemMissHandler::drop()` rather than promoted to the completed cache.
- The existing partial reader then calls `read_body()` again and receives `Ok(None)`.

The last observation is the bug-existence proof. `Ok(None)` is the same value used for a clean EOF, but the writer never wrote EOF and never called `finish()`. In the proxy path, this value can be converted to `HttpTask::Done`.

## Expected results

A streaming partial cache reader should only report clean EOF after the associated writer has actually reached EOF and completed the cache-write protocol.

More specifically:

- Dropping a `HandleMiss` before `finish()` should be observable as a failed/aborted partial write to any existing partial readers.
- `Ok(None)` from `HandleHit::read_body()` should be reserved for confirmed body completion.
- The proxy cache response state machine should not be able to convert an aborted partial cache fill into `HttpTask::Done`.

Acceptable outcomes after the writer is dropped before finish would include returning `Err`, returning an explicit aborted-writer state if the trait grows one, or switching to a recovery path before a downstream response has been committed.

## Observed results

The current in-tree streaming backend can produce this state:

```text
partial body bytes are written and visible to a reader
miss handler is dropped before EOF and before finish()
temporary object is removed from lookup-visible storage
existing partial reader keeps its receiver/body buffer
reader observes sender drop
reader returns Ok(None)
proxy state machine can treat Ok(None) as HttpTask::Done
```

For close-delimited or otherwise streaming responses, this can make a truncated cache-served body look like a successful end-of-response. For responses with `Content-Length`, a lower layer or client may detect a length mismatch, but the cache/proxy state machine still loses the more precise fact that the partial cache writer was aborted before finishing.

## Additional context

The cancellation-safety invariant I expected is:

```text
Once a partial cache body has been exposed to readers, every reader must be
able to distinguish a completed writer from an aborted writer.
```

A robust fix would likely make the partial-write protocol carry an explicit terminal state:

- represent partial write progress as `Partial(n)`, `Complete(n)`, and `Aborted(error)` rather than using writer-channel drop as an implicit EOF;
- when a miss handler is dropped before successful `finish()`, synchronously mark the partial object as aborted before removing it from lookup-visible state;
- make existing partial readers convert `Aborted` into `Err`, not `Ok(None)`;
- document that any storage backend returning `support_streaming_partial_write() == true` must distinguish clean EOF from aborted writer;
- add a storage-conformance test that creates a partial reader, drops the miss handler without `finish()`, and asserts that the reader does not return `Ok(None)` after consuming the prefix.

If the public `HandleHit::read_body() -> Result>` shape is preserved, the minimum safe rule seems to be: `Ok(None)` means confirmed body completion only; an unfinished writer disappearing while the last observed state is still `Partial(_)` should return `Err`.

Start with the streaming partial-write contract in pingora-cache/src/storage.rs and the early-unlock path in pingora-cache/src/lib.rs. Trace PartialHit::read() and MemMissHandler::drop() in pingora-cache/src/memory.rs, then follow None handling in pingora-proxy/src/proxy_cache.rs. Done means an unfinished writer is distinguishable from clean EOF and the focused memory test becomes a regression test for that behavior.

Written by the indexing model from the issue text.