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.

Why a shut read branch is not enough

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 is therefore not observed at all: propagate_reset is unreachable, and the mirrored reset that should follow the peer’s within microseconds arrives 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 differs is that the shut state is not 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 offset a silent shut state would hold it at.

That is the discriminating property, and it is why this 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.