Struct PendingQueue 

pub(crate) struct PendingQueue {

    q: VecDeque<Pending>,
    head_deadline: Option<Deadline>,
    queued_bytes: usize,
    flushed_unconfirmed: usize,
    gauge: Option<Arc<EgressGauge>>,
    config: EgressConfig,
    counters: Arc<Recorder>,
    backpressure_reported: bool,
    shape_depth: Option<QueueDepth>,
    shaper: Option<Arc<Scheduler>>,
    shape_stats: Option<Arc<ShapeRecorder>>,
    shape_side: Option<ProxySide>,
    shape_hold: Duration,
    unit_class: Class,
    shape_gate: Option<Gate>,
    shape_parked: bool,
    shape_demand: Option<Class>,
    tokens_dry: bool,
    starved_from: usize,
    shape_report: Option<ShapeReport>,
}

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

Live as a local variable in pipe_data_framed / pipe_control_mutating for the lifetime of one stream direction.

Fields

q: VecDeque<Pending>

head_deadline: Option<Deadline>

The wheel handle for q.front(). Invariant: head_deadline.is_some() == !q.is_empty().

queued_bytes: usize

flushed_unconfirmed: usize

Bytes a teardown drain handed to the transport, which nothing can confirm ever left it. See Self::unconfirmed_bytes.

gauge: Option<Arc<EgressGauge>>

The session-wide byte gauge this queue reports into, or None for a queue built outside a session — every #[cfg(test)] queue in this module, which has no session to report to.

Mirrored rather than derived: nothing can walk a session’s queues, because each one is a local variable in the pipe function that owns its stream. Every write to queued_bytes goes through Self::charge or Self::credit so the mirror cannot drift from the value it mirrors.

config: EgressConfig

counters: Arc<Recorder>

backpressure_reported: bool

Report-once latch for Push::entered_backpressure.

shape_depth: Option<QueueDepth>

The shaping profile’s per-stream depth, when this stream’s session has one and its overflow policy is Block.

None on every unshaped stream and under every non-blocking overflow: DropTail needs the read branch to stay enabled or nothing arrives to be dropped, and ResetStream is decided on the arriving unit. See Self::accepts_more.

shaper: Option<Arc<Scheduler>>

The session’s shaper, on a shaped data stream and nowhere else. None on every unshaped stream and on both control pipes, which is what makes the control pipes are never shaped structural: a control queue has no scheduler to ask, so no control frame can reach a bucket even by accident.

shape_stats: Option<Arc<ShapeRecorder>>

Where a shaped release records what it did. Carried beside the scheduler rather than reached through it because the recorder is always constructed and the scheduler is not.

shape_side: Option<ProxySide>

The side this stream’s traffic arrived on, for the per-leg and per-direction figures the recorder keeps.

Option, and deliberately not a defaulted side: one recorder serves both legs of a session, so a queue that guessed would charge real bytes to the wrong side and the figure would still look plausible. Some exactly when shape_stats is — the same builder sets both — so every reader below takes them together.

A side rather than a direction because the recorder needs the leg too, and this queue is the only thing that knows which stream it is serving. It stays the arrival side even for the release figures the proxy charges to the leg a unit leaves by: the turn-around belongs to the recorder, which is one place, rather than to every queue, which is one per stream.

shape_hold: Duration

max_hold for this stream’s shaped units: the profile’s, or the engine’s when the profile inherits it.

unit_class: Class

The class the pipe loop resolved for the unit it is currently handling. Read — not taken — by Self::push, so every push made while one framed unit is being executed carries the same class, and the next unit overwrites it.

shape_gate: Option<Gate>

The gate the head is parked on because a Discipline is holding its class back. Merged into Self::head_release so wait_release races it exactly as it races a Hold’s gate.

shape_parked: bool

Whether the shaper is currently holding the head back.

It decides which gate Self::head_release reports, and that is a spin guard, not bookkeeping. A Hold whose gate has been released makes its unit due (Pending::is_due) and the gate stays released forever; if the queue kept reporting it while the shaper was refusing the same unit, wait_release would resolve on every select! iteration and the pipe loop would spin hot until the bucket refilled. Measured before this field existed: a barrier-held object on a 0-bps class pinned a core for the whole five-second clamp and starved the test’s own task alongside it.

Once a unit is due, the only things that may wake its stream are the deadline the scheduler armed and the gate a discipline handed back.

shape_demand: Option<Class>

The class this queue has declared demand for with the scheduler. Exactly one declaration is outstanding at a time, and Drop is what guarantees it is withdrawn.

tokens_dry: bool

Edge latch for tokens_exhausted_episodes — armed when a release is refused for want of tokens, re-armed when one is granted.

starved_from: usize

The index the next starved_behind_other_class sweep starts from, so each queued unit is examined once over the queue’s whole life rather than once per wake.

shape_report: Option<ShapeReport>

What the last Self::pop_next_due wants reported, for a caller that has an exec::Reporter and an event sink. Taken, never accumulated: at most one unit is released per call.

Implementations

impl PendingQueue

pub(crate) fn new(config: EgressConfig, counters: Arc<Recorder>) -> Self

An empty queue. Allocates nothing until the first Self::push.

pub(crate) fn with_gauge(self, gauge: Arc<EgressGauge>) -> Self

Report this queue’s depth into its session’s EgressGauge.

A builder for the same reason Self::with_shape_depth is: the tests in this module build queues that belong to no session, and a required parameter would make each of them invent one. Installed by every pipe function, on every stream, shaped or not — unlike a scheduler, which only the framed data pipe installs. A gauge that only some queues reported into would answer the session has nothing left to flush while a control stream still held a deferred SUBSCRIBE_OK.

fn charge(&mut self, bytes: usize)

Add bytes to this queue’s depth and to its session’s gauge.

fn credit(&mut self, bytes: usize)

Take bytes off this queue’s depth and off its session’s gauge.

fn discarding(&self) -> bool

Whether this queue’s session has stopped flushing at teardown.

false for a queue with no gauge, which is every queue built outside a session: nothing can have asked such a queue to stop.

pub(crate) fn with_shape_depth(self, depth: Option<QueueDepth>) -> Self

Carry a shaping profile’s per-stream depth into Self::accepts_more.

A builder rather than a fourth parameter on Self::new, because every caller but one passes None and the shape of the exception is the point: exactly one construction site — the framed data pipe — has a profile to install, and the ~25 unit tests below construct queues that never had one.

Installed only under Overflow::Block, which the caller decides; this type deliberately does not name Overflow. What it does is enforce a depth, and a depth that reaches it is one the session has already decided should stop reads.

Plumbed into accepts_more rather than checked beside it because the once-per-stream backpressure latch is computed inside Self::push from accepts_more(): a stream full by QueueConfig::depth_objects but under EgressConfig::max_pending_bytes would otherwise never trip it, and Impairment{EgressQueueFull} would have no producer on a shaped stream at all.

pub(crate) fn with_shaper( self, shaper: Option<Arc<Scheduler>>, stats: Arc<ShapeRecorder>, side: ProxySide, ) -> Self

Install the session’s shaper, making this a paced queue. A builder for the same reason Self::with_shape_depth is: exactly one construction site — the framed data pipe — has a shaper, and the ~25 unit tests below build queues that never had one. Both control pipes and pipe_data_passthrough call Self::new and stop there, which is what makes the control pipes are never shaped a property of the type graph rather than of a review.

What it changes, and only this: Self::pop_next_due asks Scheduler::acquire before it yields a shaped unit, and Self::push tags what it queues. Everything else — ordering, due_at, Hold gates, the two drains, head_release’s signature — is untouched.

side travels with the recorder rather than beside it because the recorder is session-scoped and this queue is not: one ShapeRecorder is shared by the forwarding tasks of both legs, and the figures it keeps are per leg and per direction. Naming it here — at the one site that has both a profile and a stream — is what makes a downlink stall unattributable to the uplink.

It is the side this stream’s traffic arrives on, which is the only side a forwarding task is ever handed. What a release figure does with it — charge the leg the unit leaves by — is the recorder’s business, not this queue’s.

pub(crate) fn is_shaped(&self) -> bool

Whether this queue paces what it holds.

Read by exec::Engine::queue_is_busy, which is what routes a shaped stream’s units through the queue instead of writing them inline: a unit written straight to the transport never reaches Self::pop_next_due, so on a shaped stream the fast path is exactly the path that cannot be paced.

pub(crate) fn tag_unit(&mut self, class: Class)

Name the class of the unit the pipe loop is about to queue.

Set once per framed unit, immediately after classification, and read by every Self::push until the next unit overwrites it. That is why exec can queue a Delay’d object without ever naming a class: the tag is the loop’s, the push is exec’s, and neither has to know about the other.

A no-op on an unshaped queue, where nothing reads it.

pub(crate) fn take_shape_report(&mut self) -> Option<ShapeReport>

Whatever the last Self::pop_next_due wants the caller to report.

pub(crate) fn is_empty(&self) -> bool

Whether anything is waiting.

pub(crate) fn len(&self) -> usize

How many units are waiting.

Read by this module’s tests and by exec.rs’s, which assert DeferredEffects::len() == PendingQueue::len(), and on the data path by shaping admission, which needs the object count the arriving unit would be the n + 1-th of. The pipe loops themselves ask Self::accepts_more and Self::is_empty.

pub(crate) fn queued_bytes(&self) -> usize

Bytes the queue is holding, right now, still unwritten.

Not the teardown report — see Self::unconfirmed_bytes. This is the live figure the byte budget and the tests are written against.

pub(crate) fn unconfirmed_bytes(&self) -> usize

Bytes this queue cannot vouch for: everything a teardown drain handed to the transport, plus everything still queued.

What Impairment { QueuedBytesAtTeardown } carries, and the reason it is not simply Self::queued_bytes. A cancelled session’s pipe task races run_with_transport’s client.close() / relay.close(): Self::drain_ignoring_release_times can hand a 256 KiB object to quinn, have write_all return Ok because quinn buffered it, and have the connection torn down before a byte of it is on the wire. Reporting only the residue reports zero there, and the object is gone with no event at all — the one failure class this engine forbids outright: traffic is never silently gone.

So a teardown flush counts as unconfirmed, not as delivered. The event says exactly this already: “they were flushed best-effort; bytes may not have reached the peer”. A peer that did receive them gets a spurious impairment, which is the right side to err on: an over-report is a diagnosable nuisance, a silent loss is not.

Zero on every path that is not a teardown, so a stream that drains at its release times reports nothing.

pub(crate) fn config(&self) -> &EgressConfig

The engine knobs this queue was built with.

pub(crate) fn accepts_more(&self) -> bool

Whether the source may still be read from.

false once the queue holds EgressConfig::max_pending_bytes or MAX_PENDING_UNITS units — the point at which Delay stops being a latency shift and becomes backpressure — or once it reaches a shaping profile’s own depth, when one was installed by Self::with_shape_depth. A bool, deliberately: it is hoisted out of the select! and holds no borrow.

Read from, not observed. This gates recv.read, which is the only call that consumes bytes and therefore the only one that grants the peer flow-control credit. It does not gate whether the pipe loop watches its source: false swaps the read for RecvStream::received_reset, so a peer’s RESET_STREAM is mirrored at once however full this queue is. Gating the observation on this was the defect — a dry bucket under Overflow::Block held the read shut for max_hold (30 s by default) and the reset went unseen for the whole of it.

Both terms are per stream, and both imply a non-empty deque, so head_release() stays Some and the release branch stays live. A shared cross-class budget must never reach this function: an empty own queue with a shared budget exhausted disables the read branch and the release branch, and the stream deadlocks until session teardown.

pub(crate) fn head_release(&self) -> Option<Release>

What the head is waiting for, as an owned clone — see Release.

&self, and it debits nothing. It reports the deadline the scheduler last computed and the gate it last parked on; the decision itself belongs to Self::pop_next_due, which is &mut self and runs exactly once per released unit. This function is called once per select! iteration from a hoisted local, so a debit here would be charged per read-wake rather than per byte written.

The FIN drain is on pop_next_due’s path, and this sentence used to deny it. drain_honouring_release_times — the read arm’s None branch — loops pop_next_due, debits the bucket, and applies the clamp and the expiry, which is why it takes an on_shape callback and reports them. The control pipes really are exempt, but by construction rather than by call graph: they install no Scheduler at all, so shaping_grants_head short-circuits on shaper: None and no control frame can reach a bucket even by accident. The only drain that is genuinely off the seam is Self::drain_ignoring_release_times, which pops the front directly and consults nothing.

fn tail_expected_at(&self) -> Option<Instant>

When the queue expects to have written the last unit in it, or None if it is empty.

None rather than Instant::now(): the estimate exists only to say what a unit waits behind, and on an empty queue there is nothing ahead. Folding now in there would quietly shift every unit’s reported release time forward to its push instant.

fn arm_head(&mut self)

Register the head with the release wheel.

Armed at the head’s due_at, never its expected_at: the head has nothing ahead of it, so its own deadline is when it may go, and arming on the estimate would re-introduce the successor stall the module doc describes (a unit clamped behind a Hold would sleep to that hold’s ceiling even after it became the head).

The only caller of release_timer::arm_at besides Self::push’s use of it through here. arm_at starts nothing when the head is already due, so a queue of undelayed units never constructs the wheel.

fn arm_head_at(&mut self, at: Instant)

Re-arm the head at an instant the scheduler named, rather than at the head’s own due_at. The head is due on its own account — that is how it reached the shaper — so re-arming at due_at would give an already-fired deadline and spin the release branch hot until the bucket refilled. This is what makes head_release() the deadline the scheduler last computed: it reads back exactly what was armed here.

pub(crate) fn push(&mut self, unit: Pending) -> Push

Queue a unit behind everything already waiting.

due_at is not touched. The unit keeps the deadline it was built with, because that deadline is its own: a Delay’s arrived_at + by, a Hold’s max_hold ceiling, or now for a unit queued purely for ordering. Wire order is held by the deque — Self::pop_next_due only ever considers the front — so nothing has to be added to a successor’s deadline to keep it behind its predecessor, and adding something is exactly what stranded a stream behind a released Hold (module doc). What is computed here is Pending::expected_at, the queue’s estimate of when this unit will reach the wire: max(due_at, tail.expected_at, now) on a non-empty queue, and due_at on an empty one. It is what Push::release_at returns, so Effect::Queued { release_at } still says no earlier than everything ahead of you, and it is what Self::record_release measures lateness against, so a Pass queued behind a 300 ms Delay is not logged as a 300 ms release error.

A unit that is due now on an empty queue should not be pushed at all — the pipe loop writes it inline in the read arm, which is today’s code and today’s cost.

pub(crate) fn pop_next_due(&mut self, now: Instant) -> Option<Pending>

Pop the head if it is due at now, re-arming the new head.

The front and only the front. That single fact is the whole ordering guarantee: a unit whose own deadline passed long ago is still not popped while anything is queued in front of it.

Call it in a while let loop: pop_next_due yields one unit at a time and allocates nothing, where a Vec-returning pop_due would allocate on every release wake. Looping matters as much as it reads: one wake on a Hold’s gate has to drain the held unit and every unit behind it that is due on its own account, or releasing a gate resumes one object and strands the rest.

The release seam

On a shaped queue — and only there — a unit that is due on its own account is then put to Scheduler::acquire, which debits its class’s token bucket and applies the Discipline. This function is where that happens because it is &mut self, it is the sole ordering authority, and it is called exactly once per released unit; head_release is none of those things.

A refusal is not an error and costs nothing: the head deadline is re-armed at whatever the scheduler named (clamped by the unit’s own max_hold), the gate a discipline handed back is stored for head_release, and None is returned exactly as a not-yet-due head returns it. The caller’s loop is unchanged.

fn shaping_grants_head(&mut self, now: Instant) -> bool

Whether the shaper lets the head go at now, arming whatever it must wait for when it does not.

true on every unshaped queue and for every unit the shaper does not own — a terminal, an elided ordering slot, a header — so the only thing a bucket can ever hold back is bytes.

fn note_tokens_dry(&mut self, class: Class)

Charge one episode of this class’s own bucket being dry, on the edge only.

Shared by the three refusals that are the bucket’s rather than a discipline’s, so the edge latch cannot be re-armed on one of them and forgotten on another. Re-armed by a grant, in shaping_grants_head.

fn park_head(&mut self, at: Instant, tag: ShapeTag, gate: Option<Gate>)

Park the head until at — or until gate opens, when a discipline is what is holding it — and charge everything queued behind a unit of another class.

fn note_head_starved(&mut self)

Charge the head once against starved_behind_other_class.

fn note_starved_behind(&mut self, head_class: Class)

Charge starved_behind_other_class for every queued unit that is waiting behind a unit of a different class.

Once per unit over the queue’s whole life, not once per wake: the sweep resumes from where the last one stopped and the cursor follows the front as units are popped, so the total work is linear in units queued rather than quadratic in wakes.

Deliberately separate from tokens_exhausted_episodes. Head-gating makes configured shaping and head-of-line blocking look identical from outside, and this is the number that tells them apart.

fn expire_head(&mut self, expiry: Expiry) -> bool

The head outlived its clamp. Deliver it, or abandon the stream.

Returns whether the caller may pop the head. Under Expiry::Deliver it may — that is the whole of “clamps and delivers”, and it is why objects_expired is zero by default.

fn note_unshapeable_seen(&self, unit: &Pending)

Account for a queued unit no rule could see: the left-hand side of the conservation identity, for bytes the classifier never met. A stream header, an oversized object’s passthrough chunk and a bypassed stream’s tail carry no ObjectMeta, so note_object_seen is never reached for them and bytes_shaped would not count them. They are still bytes the shaper handled — it queued them, it ordered them, and it wrote them — so leaving them out would make bytes_shaped mean “bytes a rule saw” while ShapeStats::bytes_shaped promises every byte it accounted for, and the unshapeable row would have nothing to be the other half of.

Here and not at the call site, which is what makes the identity structural: push is the one place an unshapeable unit enters, this reads the same unit.len() Self::note_delivered will charge when it leaves, and both are gated on the same ShapeTag. session.rs cannot get it wrong because session.rs is not asked — the object arm and the unshown arm share one exec entry point, and only the tag tells them apart.

Zero-length units — an elided ordering slot, a queued reset — add nothing on either side.

fn note_delivered(&self, unit: &Pending)

Charge one released unit to its class.

Class::Unshapeable is charged like any other row, and only because Self::note_unshapeable_seen now counts the same bytes into bytes_shaped on the way in: both terms move together or neither does, and this is the unit in which they moved. The consequence worth stating: unshapeable bytes are not paced. shaping_grants_head never asks a bucket about them — they carry no class a bucket is configured for — so an object too large for the framer to buffer bypasses every rate, and a class rate can be exceeded by exactly one oversized object. That is what “unshapeable” means; the row named for it is where the figure is reported, and it is deliberately a separate row from default_class so that no rule claimed this unit and “no rule could have” stay two answers.

The row alone does not say whose rate went, which is the question an author with a 500 kbps video cap and a large initial segment is actually asking, so the pipe loop pairs it with Impairment{ShapeUnpacedObject} naming the class the stream’s classified units are charged to.

fn sync_shape_demand(&mut self)

Tell the scheduler which class this queue is holding an unreleased head of, if any.

Demand, not depth. One declaration per queue, replaced whenever the head’s class changes and withdrawn when the queue empties — which includes Self::clear, the teardown drain, and Drop. A declaration that outlived its stream would starve a lower-priority class for the rest of the session, and the failure would look like a hang rather than like a leak.

pub(crate) fn record_release(&self, unit: &Pending, now: Instant)

Sample one deferred release’s lateness onto the session counters.

Call it only from the release branch, once per unit pop_next_due yields, before writing. Units written inline in the read arm are not releases and would dilute the distribution with zeros; drains ignore release times by design and recording their lateness would report teardown as a timing failure. Neither drain in this file calls it. Measured against Pending::expected_at, not Pending::due_at: the question is did the engine release when it said it would, and what it said was Effect::Queued { release_at }. Sampling due_at instead would log the queueing delay of every unit behind a Delay as a timer error.

saturating_duration_since because a unit released early — a Hold’s gate opening long before its ceiling, which is the normal case — is a zero, not a negative.

pub(crate) fn clear(&mut self)

Forget everything queued. Used after a terminal fires — the destination is reset, so nothing behind it can be written.

Dropping the Releases abandons their wheel slots, which the wheel prunes at their own instants or at a sweep, without waking anything.

pub(crate) async fn drain_ignoring_release_times<S: EgressSink>( &mut self, send: &mut S, ) -> DrainOutcome

Write everything queued now, in order, ignoring release times.

The teardown path: delivered late beats lost silently. Best effort by construction — a write failure stops the drain and leaves the failing unit and everything behind it queued, so Self::queued_bytes reports exactly what did not reach the transport.

Every byte it does hand over is added to Self::unconfirmed_bytes, because a write_all that returns Ok into a connection another task is about to close() is not a delivery. That is the whole of the teardown-race loss: read Self::unconfirmed_bytes after this, never Self::queued_bytes, on any path that runs because the session is going down.

Consults no bucket. This is the teardown drain, and teardown bypasses the pacer entirely: it pops the front directly rather than through Self::pop_next_due, so a token bucket can never gate a mirrored reset. That is what keeps the eleven data-then-teardown ordering tests independent of any configured rate.

And therefore reports no shaping, deliberately. Every other release path takes a ShapeReport out of the queue and hands it to a reporter, because a clamp or an expiry that nothing says happened is indistinguishable from a profile that did nothing. This one is the exception, and it is an exception because it applies no shaping: Self::pop_next_due is the only producer of a ShapeReport and this function never calls it, so there is nothing to drain rather than something dropped. The slot is empty on entry as well — both callers that can reach it through a shaped queue (drain_honouring_release_times’s cancel arm and release_due_units) drain the report before the pop that produced it is written.

Known residue, stated rather than fixed: popping the front directly also skips the shape_parked / shape_gate bookkeeping Self::pop_next_due maintains, so after this drain a Self::head_release would describe a park belonging to a unit that is gone. Nothing reads it — every caller returns immediately, the queue is empty or cleared, and head_release answers None on an empty deque — so there is no failure case here to fix. It is written down because “unreachable today” is a property of the callers, not of this function.

Trait Implementations

impl Debug for PendingQueue

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Drop for PendingQueue

Withdraw whatever demand this queue still holds.

Drop rather than a call on each teardown path, and for the same reason StreamGuard is: a forwarding task can end in at least five different ways, any hand-written list of them is only as complete as its reader, and Drop additionally covers a ? return, a panicking task, and JoinSet::shutdown dropping the future wholesale. A leaked declaration is not a loud failure — it is a different stream stalling to max_hold, on a class the reader was not looking at.

fn drop(&mut self)

Executes the destructor for this type.

fn pin_drop(self: Pin<&mut Self>)

🔬This is a nightly-only experimental API. (pin_ergonomics)

Execute the destructor for this type, but different to Drop::drop, it requires self to be pinned.