Module egress 

The per-stream deferred-write deque — the engine behind Delay and Hold.

A PendingQueue is a local variable in the pipe function that already owns both halves of the stream. No task, no channel, no Semaphore: the reader keeps &mut send, so a mirrored peer reset still leaves the destination after the bytes already written, which is what proxy_reset.rs::upstream_reset_reaches_the_client_as_a_reset_with_the_same_code (data, then reset) asserts. A writer-task design reverses that ordering and cannot be made to pass it.

The queue is empty and non-allocating unless a timing action is used — VecDeque::new does not allocate — so a session that only passes, replaces or drops never touches this module’s hot path and never links crate::release_timer into any executed code.

The shape the pipe loop takes

let can_read     = pending.accepts_more();   // byte + unit budget, a bool
let head_release = pending.head_release();   // Option<Release>, owned clone

tokio::select! {
    // `can_read` chooses *inside* observe_source: true polls
    // `recv.read`, false polls `recv.received_reset()`, which consumes
    // no bytes. The source is never unobserved — see `observe_source`.
    source = observe_source(&mut recv, &mut buf, can_read, watch) => { /* enqueue-or-write-now */ }
    _ = egress::wait_release(head_release.clone(), &ctx.cancel), if head_release.is_some() => {
        let now = Instant::now();
        while let Some(unit) = pending.pop_next_due(now) {
            pending.record_release(&unit, now);
            egress::write_unit(unit, &mut send).await?;
        }
    }
    _ = ctx.cancel.cancelled() => { /* framer.finish() flush, then drain, then return */ }
}

Both branch expressions are free of any borrow of pending: one is a bool, the other an owned Release (one Arc clone plus one token clone). That is a rule, not a compile error — a branch future holding &mut pending whose winning body then calls pending.pop_next_due() does in fact compile, because select! drops the branch-future tuple before the handler runs. Keeping the expressions borrow-free is what lets wait_release be a free function shared by the release branch and by drain_honouring_release_times, and lets a later edit add a fourth branch without re-arranging the other three.

Why [tokio_util::sync::CancellationToken] and not a Notify or a waker registration: it is level-triggered. The select! may create and drop a fresh cancelled() future on every iteration with no lost-wakeup window and no re-registration, whereas a Notify-based wheel would need a global-mutex round trip per read chunk to re-register. if head_release.is_some() matters for the same reason from the other side: tokio does not evaluate a branch’s async expression when its precondition is false, so an empty queue costs one Option::is_some().

Cancellation

Every wait in this module is a select! branch, never an arm body: an arm body is not preemptible, so an inline sleep there starves the cancel branch and a Hold on a gate nobody releases would pin session teardown for up to EgressConfig::max_hold (30 s by default). That applies to the two release-honouring drains as much as to the pipe loop, which is why drain_honouring_release_times is written as a biased race against cancellation with PendingQueue::drain_ignoring_release_times as its fallback.

Ordering is the queue’s; readiness is the unit’s

The two are separate properties, and conflating them is what the first shape of this module got wrong.

• Ordering belongs to the deque. PendingQueue::pop_next_due only ever looks at the front, so nothing can be written before everything ahead of it has been. That invariant needs no arithmetic at all, and it is absolute: anything else silently corrupts MoQT stream semantics.
• Readiness belongs to the unit. A Pending is due when Pending::due_at has passed or its own Hold gate is open — see Pending::is_due. Nothing that happens to a unit ahead of it may rewrite that.

So PendingQueue::push does not clamp due_at against the tail. An earlier shape did, and it turned a Hold’s EgressConfig::max_hold ceiling into a deadline conferred on every successor: releasing the gate woke only the unit carrying it, and objects queued behind — including the stream’s FIN, which waits for the queue to drain — sat until max_hold (30 s by default). Worse, a Delay’s own deadline was overwritten by the ceiling of a Hold in front of it, so the deadline Action::Delay documents (“arrived_at + by, a deadline, not a spacing”) was not the one the engine honoured.

What the clamp was for is still wanted, and it survives as a separate field: Pending::expected_at is max(due_at, tail.expected_at, now), the queue’s estimate of when the unit will actually reach the wire given everything ahead of it. It is what Push::release_at reports as Effect::Queued { release_at } and what PendingQueue::record_release measures lateness against, so a unit queued behind a 300 ms delay is not logged as a 300 ms timer error. It never gates anything.

Bound

PendingQueue::accepts_more is false once the queue holds EgressConfig::max_pending_bytes (1 MiB default) or MAX_PENDING_UNITS units. The pipe loop stops calling recv.read on false, so the queue cannot grow past that plus one in-flight read chunk. It does not stop observing the source: it swaps the read for RecvStream::received_reset, which consumes nothing, so the bound here is unchanged and a peer’s RESET_STREAM is still seen at once. The unit cap is not redundant with the byte cap: an elided unit holds an ordering slot and carries zero bytes, so a hook that elides every object behind a never-released Hold would otherwise grow the deque without bound at zero bytes. Crossing either cap is reported once per stream — see Push::entered_backpressure.

Structs

CloserInner 🔒

Deferral 🔒

A resolved release deadline, and whether EgressConfig::max_hold cut it short.

EgressGauge 🔒

How many bytes one session’s egress queues are holding, right now.

Pending 🔒

One unit of traffic waiting for its release time.

PendingQueue 🔒

A per-stream FIFO of units waiting for their release times.

Push 🔒

What PendingQueue::push did.

Release 🔒

What the head of a queue is waiting for. Owned, cheap to clone.

SessionCloser 🔒

Where crate::action::Action::CloseSession lands.

ShapeTag 🔒

What the shaper attached to one queued unit, on a shaped stream.

Enums

CloseOrigin 🔒

Who asked for a session close.

DrainOutcome 🔒

How a drain ended.

EgressError 🔒

What writing to a destination can fail with.

Item 🔒

What a queued unit does when it is written.

ShapeReport 🔒

What a shaped release wants the pipe loop to report.

Terminal 🔒

A positional stream ending.

Written 🔒

What handing one unit to the transport did.

Constants

DEFAULT_CLOSE 🔒

The default (code, reason) run_with_transport closes with when no hook asked for anything else — today’s hard-coded pair.

MAX_PENDING_UNITS 🔒

Hard cap on queued units, independent of the byte budget.

Traits

EgressSink 🔒

The write half this module drives.

Functions

defer_by 🔒

Resolve Delay { by } against EgressConfig::max_hold.

drain_honouring_release_times 🔒

Write everything queued at its release time, racing cancellation.

hold_ceiling 🔒

The ceiling on an crate::action::Action::Hold: a gate nobody releases is still written at arrived_at + max_hold, so a hook cannot pin a stream open forever.

wait_release 🔒

Resolve when the head unit may be written, or when the session is torn down — whichever comes first.

write_unit 🔒

Write one popped unit.