Function observe_source 

async fn observe_source(
    recv: &mut PeekedRecv,
    buf: &mut [u8],
    can_read: bool,
    reset_observable: bool,
) -> Source

Observe the source stream, whatever the egress queue is doing.

The defect this exists to close

Both queueing pipe loops gate their read branch on PendingQueue::accepts_more(), and that is the backpressure mechanism: when it is false tokio does not evaluate the branch’s expression, so recv.read is not polled and nothing is consumed. Under Overflow::Block a dry bucket holds the queue at depth_objects indefinitely, so the gate stays shut for as long as max_hold — 30 s in the shipped default posture.

A peer’s RESET_STREAM surfaces only as Err from recv.read. With the read branch shut it was therefore not observed at all: propagate_reset was unreachable, and the mirrored reset that should follow the peer’s within microseconds arrived up to max_hold late. The other three branches cannot cover it — StopWatcher watches the destination’s stopped(), the release branch watches this proxy’s own clock, and cancel is session teardown.

Why this does not delete Overflow::Block

can_read still gates recv.read, which is the only call that consumes bytes. Nothing about the queue’s depth, the admission decision, or the once-per-stream backpressure latch moves. What changes is that the shut state is no longer silent: instead of parking on nothing, the loop parks on RecvStream::received_reset, which reads no bytes and therefore grants no MAX_STREAM_DATA credit. The peer stays blocked at exactly the same offset it was blocked at before.

That is the discriminating property, and it is why the fix is not “poll recv.read anyway and park the chunk”: a look-ahead slot consumes a chunk, and — worse — it only re-opens when the queue drains, so under a dry bucket the next reset waits out max_hold all the same.

Cancel safety

Every path awaits exactly one future and does nothing before it: RecvStream::read and RecvStream::received_reset are both cancel-safe, and pending() never completes. Dropping this future — which select! does on every iteration another branch wins — loses nothing.