Skip to main content

vibeio_http/h3/qpack/decoder/
mod.rs

1//! QPACK decoder (RFC 9204 Sections 2.2, 4.3 and 4.5).
2//!
3//! Consumption: the HTTP/3 layer drives the decoder per connection (it feeds
4//! encoder stream data and decoded field sections, and drains the decoder
5//! stream); until that lands, the whole module is dead in non-test builds,
6//! which is why `dead_code` is expected here. It errors again once the
7//! decoder is used, reminding us to remove the expectation.
8//!
9//! The decoder materializes the shared dynamic table (Section 4.2) from the
10//! encoder stream (Section 4.3): every instruction is parsed and mirrored as
11//! a table insertion or capacity change. Field sections (Section 4.5) are
12//! decoded against the table; a section whose Required Insert Count exceeds
13//! the decoder's insert count is buffered as blocked (Section 2.2.1) until a
14//! later encoder stream update makes it decodable.
15//!
16//! The decoder emits decoder stream instructions (Section 4.4): a Section
17//! Acknowledgment for every decoded field section with a positive Required
18//! Insert Count (Section 2.2.2.1), a Stream Cancellation for abandoned or
19//! timed-out blocked streams (Section 2.2.2.2), and coalesced Insert Count
20//! Increment instructions (Section 2.2.2.3).
21//!
22//! Validation is strict: malformed instructions are `QPACK_ENCODER_STREAM_
23//! ERROR`, malformed field sections are `QPACK_DECOMPRESSION_FAILED`, a
24//! Required Insert Count that does not equal the largest referenced absolute
25//! index plus one is rejected (Sections 2.1.2 and 2.2.1), evictions that
26//! touch entries with an absolute index at or above the Known Received Count
27//! are rejected (Sections 2.1.1 and 3.2.2), and field sections that push a
28//! stream's cumulative decoded size over the advertised
29//! `SETTINGS_MAX_FIELD_SECTION_SIZE` are rejected (RFC 9114 Section
30//! 7.2.4.1).
31#![expect(dead_code)]
32
33use std::collections::VecDeque;
34
35use bytes::Bytes;
36
37use crate::h3::qpack::error::QpackError;
38use crate::h3::qpack::static_table;
39use crate::h3::qpack::table::DynamicTable;
40use crate::hpack::{huffman, integer, HpackError};
41
42/// `1` + 7-bit stream ID: Section Acknowledgment (RFC 9204 4.4.1).
43const SECTION_ACK: u8 = 0x80;
44/// `01` + 6-bit stream ID: Stream Cancellation (RFC 9204 4.4.2).
45const STREAM_CANCELLATION: u8 = 0x40;
46/// `00` + 6-bit increment: Insert Count Increment (RFC 9204 4.4.3).
47const INSERT_COUNT_INCREMENT: u8 = 0x00;
48
49// Encoder instruction patterns (RFC 9204 4.3), mirrored from the encoder.
50/// `001` + 5-bit capacity: Set Dynamic Table Capacity (4.3.1).
51const SET_CAPACITY: u8 = 0b0010_0000;
52/// `1 T` + 6-bit name index: Insert with Name Reference (4.3.2).
53const INSERT_WITH_NAME_REF: u8 = 0b1000_0000;
54/// `01` + H + 5-bit name length: Insert with Literal Name (4.3.3).
55const INSERT_WITH_LITERAL_NAME: u8 = 0b0100_0000;
56/// `000` + 5-bit relative index: Duplicate (4.3.4).
57const DUPLICATE: u8 = 0b0000_0000;
58
59// Field line patterns (RFC 9204 4.5), mirrored from the encoder.
60/// `1 T` + 6-bit index: Indexed Field Line (4.5.2).
61const INDEXED: u8 = 0b1000_0000;
62/// `0001` + 4-bit post-Base index: Indexed Field Line with Post-Base Index
63/// (4.5.3).
64const INDEXED_POST_BASE: u8 = 0b0001_0000;
65/// `01 N T` + 4-bit name index: Literal Field Line with Name Reference
66/// (4.5.4).
67const LITERAL_NAME_REF: u8 = 0b0100_0000;
68/// `0000 N` + 3-bit post-Base name index: Literal Field Line with Post-Base
69/// Name Reference (4.5.5).
70const LITERAL_POST_BASE_NAME_REF: u8 = 0b0000_0000;
71/// `001 N` + H + 3-bit name length: Literal Field Line with Literal Name
72/// (4.5.6).
73const LITERAL_LITERAL_NAME: u8 = 0b0010_0000;
74
75/// A field section that was buffered as blocked and has since been decoded
76/// after an encoder stream update.
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub struct UnblockedSection {
79    /// The stream the encoded field section was received on.
80    pub stream_id: u64,
81    /// The decoded header list.
82    pub headers: Vec<(Bytes, Bytes)>,
83}
84
85/// A field section buffered because its Required Insert Count had not been
86/// reached yet (RFC 9204 Section 2.2.1).
87#[derive(Debug)]
88struct BlockedSection {
89    stream_id: u64,
90    buf: Bytes,
91    ric: u64,
92    since: u64,
93}
94
95/// QPACK decoder: dynamic table mirror and field section decoder.
96#[derive(Debug)]
97pub struct Decoder {
98    dynamic: DynamicTable,
99    /// The maximum dynamic table capacity advertised by this decoder in
100    /// SETTINGS_QPACK_MAX_TABLE_CAPACITY (RFC 9204 Section 3.2.3).
101    max_capacity: u64,
102    /// Total insertions and duplications received on the encoder stream.
103    /// Part of the field section prefix decoding context.
104    ///
105    /// The Known Received Count, which rules evictability and the Insert
106    /// Count Increment instruction, is tracked separately in
107    /// [`Decoder::known_received`] (RFC 9204 Section 2.1.4).
108    ///
109    /// Invariant: `known_received <= inserted`.
110    known_received: u64,
111    /// Blocked field sections, in arrival order.
112    blocked: VecDeque<BlockedSection>,
113    /// Number of blocked sections per stream. Kept separately so admission
114    /// does not rescan every buffered section (or allocate a temporary list)
115    /// for every field section received while the table is catching up.
116    blocked_by_stream: Vec<(u64, usize)>,
117    /// Decoded field-section size per stream, summed so the
118    /// `SETTINGS_MAX_FIELD_SECTION_SIZE` budget applies across a stream's
119    /// field sections (request headers, trailers) like
120    /// `SETTINGS_MAX_HEADER_LIST_SIZE` does for HTTP/2 — not per section.
121    ///
122    /// A stream's budget is only charged when a section is actually
123    /// decoded, so a section buffered as blocked is charged when it is
124    /// unblocked. Entries are dropped when the stream finishes, is reset,
125    /// or is abandoned ([`Decoder::stream_finished`],
126    /// [`Decoder::stream_cancelled`], [`Decoder::expire_blocked`]); QUIC
127    /// stream IDs are never reused, so a stale entry could not affect
128    /// another stream anyway.
129    section_size_by_stream: Vec<(u64, usize)>,
130    /// The maximum number of streams that may be blocked at once,
131    /// SETTINGS_QPACK_BLOCKED_STREAMS (RFC 9204 Section 5).
132    max_blocked_streams: usize,
133    /// Cap on the total size of a decoded field section (the sum of the
134    /// lengths of the names and values of its field lines), the locally
135    /// advertised `SETTINGS_MAX_FIELD_SECTION_SIZE` (RFC 9114 Section
136    /// 7.2.4.1). Exceeding it is `QPACK_DECOMPRESSION_FAILED`.
137    max_field_section_size: usize,
138    /// Decoder stream instructions awaiting transmission.
139    decoder_stream: Vec<u8>,
140    /// Bytes of the peer's encoder stream received so far but not yet forming
141    /// a complete instruction. QPACK encoder-stream instructions carry
142    /// variable-length strings and can span the arbitrary chunk boundaries of
143    /// the underlying QUIC stream, so partial instructions are buffered here
144    /// until the rest arrives (RFC 9204 Section 4.3) instead of being treated
145    /// as a stream error.
146    encoder_stream_pending: Vec<u8>,
147    /// Bytes of the peer's decoder stream received so far but not yet forming
148    /// a complete instruction. The same chunk-boundary buffering as
149    /// `encoder_stream_pending` (RFC 9204 Section 4.4).
150    decoder_stream_pending: Vec<u8>,
151}
152
153/// Upper bound on the buffered prefix of either peer stream. A complete
154/// decoder-stream instruction is a single prefixed integer of at most 10
155/// bytes, so a buffered prefix far larger than this can never become one.
156const MAX_DECODER_STREAM_PENDING: usize = 64;
157
158/// Byte length of a string literal whose length integer starts at the
159/// beginning of `buf` with `prefix_bits` (the Huffman/control bit occupies the
160/// high bit of that prefix, so the integer itself uses `prefix_bits - 1`
161/// bits). Returns `None` when `buf` is too short to hold the length integer
162/// or its advertised content.
163fn string_len(buf: &[u8], prefix_bits: u8) -> Option<usize> {
164    let int_prefix = prefix_bits - 1;
165    let int_len = integer::encoded_len(buf, int_prefix)?;
166    // `decode` consumes `buf[0]` as the header, then continuation octets from
167    // `off` onward, so `off` must point past the header byte.
168    let mut off = 1;
169    let len = integer::decode(buf, &mut off, int_prefix, buf[0]).ok()?;
170    let content = usize::try_from(len).ok()?;
171    let total = int_len.checked_add(content)?;
172    if buf.len() < total {
173        return None;
174    }
175    Some(total)
176}
177
178/// Byte length of one encoder-stream instruction (RFC 9204 Section 4.3)
179/// starting at the beginning of `buf`, or `None` when `buf` is too short to
180/// contain the whole instruction.
181fn encoder_instruction_len(buf: &[u8]) -> Option<usize> {
182    let header = *buf.first()?;
183    match header & 0xC0 {
184        // Insert with Name Reference (4.3.2): relative/static index (6-bit
185        // prefix) followed by the value string.
186        0x80 | 0xC0 => {
187            let index_len = integer::encoded_len(buf, 6)?;
188            let value_len = string_len(buf.get(index_len..)?, 8)?;
189            Some(index_len + value_len)
190        }
191        // Insert with Literal Name (4.3.3): name string (the instruction's
192        // first byte doubles as the name-length prefix, 6-bit prefix) followed
193        // by the value string (8-bit prefix).
194        0x40 => {
195            let name_len = string_len(buf, 6)?;
196            let value_len = string_len(buf.get(name_len..)?, 8)?;
197            Some(name_len + value_len)
198        }
199        // `00`: Set Dynamic Table Capacity (4.3.1) or Duplicate (4.3.4), each
200        // a 5-bit prefixed integer.
201        _ => integer::encoded_len(buf, 5),
202    }
203}
204
205/// Byte length of one decoder-stream instruction (RFC 9204 Section 4.4)
206/// starting at the beginning of `buf`, or `None` when `buf` is too short.
207fn decoder_instruction_len(buf: &[u8]) -> Option<usize> {
208    let header = *buf.first()?;
209    let prefix_bits: u8 = if header & 0x80 != 0 { 7 } else { 6 };
210    integer::encoded_len(buf, prefix_bits)
211}
212
213impl Decoder {
214    /// Creates a decoder that advertised `max_capacity` in
215    /// SETTINGS_QPACK_MAX_TABLE_CAPACITY and `max_blocked_streams` in
216    /// SETTINGS_QPACK_BLOCKED_STREAMS.
217    #[inline]
218    pub fn new(max_capacity: u64, max_blocked_streams: usize) -> Self {
219        Self {
220            dynamic: DynamicTable::new(0),
221            max_capacity,
222            known_received: 0,
223            blocked: VecDeque::new(),
224            blocked_by_stream: Vec::new(),
225            section_size_by_stream: Vec::new(),
226            max_blocked_streams,
227            max_field_section_size: usize::MAX,
228            decoder_stream: Vec::new(),
229            encoder_stream_pending: Vec::new(),
230            decoder_stream_pending: Vec::new(),
231        }
232    }
233
234    /// Sets the maximum size of a decoded field section: the locally
235    /// advertised `SETTINGS_MAX_FIELD_SECTION_SIZE` (RFC 9114 Section
236    /// 7.2.4.1). The budget is per stream and accumulates across its field
237    /// sections (request headers and trailers); a section that pushes a
238    /// stream's cumulative name and value octets over this is rejected
239    /// with `QPACK_DECOMPRESSION_FAILED` (RFC 9204 Section 4.5).
240    #[inline]
241    pub fn set_max_field_section_size(&mut self, size: usize) {
242        self.max_field_section_size = size;
243    }
244
245    /// The total number of insertions and duplications materialized from the
246    /// encoder stream so far; the decoder's Insert Count (RFC 9204
247    /// Section 2.1.1).
248    #[inline]
249    pub fn inserted(&self) -> u64 {
250        self.dynamic.inserted()
251    }
252
253    /// The Known Received Count: insertions and duplications the decoder has
254    /// acknowledged or incremented (RFC 9204 Section 2.1.4).
255    #[inline]
256    pub fn known_received(&self) -> u64 {
257        self.known_received
258    }
259
260    /// The number of field sections currently buffered as blocked.
261    #[inline]
262    pub fn pending_blocked(&self) -> usize {
263        self.blocked.len()
264    }
265
266    /// Takes the accumulated decoder stream instructions.
267    #[inline]
268    pub fn take_decoder_stream(&mut self) -> Bytes {
269        Bytes::from(std::mem::take(&mut self.decoder_stream))
270    }
271
272    /// Parses the peer's QPACK decoder stream instructions (RFC 9204
273    /// Section 4.4): Section Acknowledgments, Stream Cancellations, and
274    /// Insert Count Increments.
275    ///
276    /// An Insert Count Increment with a zero value is a decoder stream
277    /// error (Section 4.4.3); any malformed instruction is too. The
278    /// instructions are not otherwise acted upon: this decoder emits its own
279    /// decoder-stream instructions and never tracks the peer's
280    /// acknowledgements, so it only needs to validate what the peer sends.
281    #[inline]
282    pub fn feed_decoder_stream(&mut self, buf: &[u8]) -> Result<(), QpackError> {
283        self.decoder_stream_pending.extend_from_slice(buf);
284        // A decoder-stream instruction is a single prefixed integer whose
285        // length is known once enough bytes arrive; parse only complete
286        // instructions and buffer any trailing partial one (RFC 9204 4.4). The
287        // QUIC stream beneath can deliver an instruction split across chunks.
288        if self.decoder_stream_pending.len() > MAX_DECODER_STREAM_PENDING {
289            return Err(QpackError::DecoderStream);
290        }
291        let mut consumed = 0;
292        while consumed < self.decoder_stream_pending.len() {
293            let Some(len) = decoder_instruction_len(&self.decoder_stream_pending[consumed..])
294            else {
295                break;
296            };
297            if consumed + len > self.decoder_stream_pending.len() {
298                break;
299            }
300            let instr = &self.decoder_stream_pending[consumed..consumed + len];
301            let mut off = 0;
302            let header = instr[0];
303            if header & 0x80 != 0 {
304                // `1` + 7-bit stream ID: Section Acknowledgment (4.4.1).
305                integer::decode(instr, &mut off, 7, header).map_err(dec_stream_err)?;
306            } else if header & 0x40 != 0 {
307                // `01` + 6-bit stream ID: Stream Cancellation (4.4.2).
308                integer::decode(instr, &mut off, 6, header).map_err(dec_stream_err)?;
309            } else {
310                // `00` + 6-bit increment: Insert Count Increment (4.4.3). A
311                // zero increment is forbidden.
312                let increment =
313                    integer::decode(instr, &mut off, 6, header).map_err(dec_stream_err)?;
314                if increment == 0 {
315                    return Err(QpackError::DecoderStream);
316                }
317            }
318            consumed += len;
319        }
320        self.decoder_stream_pending.drain(..consumed);
321        Ok(())
322    }
323
324    /// Processes the encoder stream instructions in `buf`, materializing
325    /// dynamic table updates.
326    ///
327    /// Returns the field sections that were blocked and can now be decoded,
328    /// in arrival order. The Section Acknowledgment for each is queued in
329    /// the decoder stream, together with a coalesced Insert Count Increment
330    /// when the table grew beyond the acknowledged count.
331    #[inline]
332    pub fn feed_encoder_stream(&mut self, buf: &[u8]) -> Result<Vec<UnblockedSection>, QpackError> {
333        self.encoder_stream_pending.extend_from_slice(buf);
334        // Buffer partial instructions: an encoder-stream instruction can carry
335        // variable-length strings and may arrive split across the arbitrary
336        // chunk boundaries of the underlying QUIC stream, so only complete
337        // instructions are processed (RFC 9204 Section 4.3). A complete
338        // instruction is bounded by the dynamic table capacity, so a much
339        // larger buffered prefix can never become one and is rejected to bound
340        // memory against a peer that streams continuation bytes.
341        let cap = (self.max_capacity as usize).saturating_add(1024);
342        if self.encoder_stream_pending.len() > cap {
343            return Err(QpackError::EncoderStream);
344        }
345        let mut consumed = 0;
346        while consumed < self.encoder_stream_pending.len() {
347            let Some(len) = encoder_instruction_len(&self.encoder_stream_pending[consumed..])
348            else {
349                break;
350            };
351            if consumed + len > self.encoder_stream_pending.len() {
352                break;
353            }
354            // Copy the complete instruction so it can be parsed while `self`
355            // is mutably borrowed by the insert below (no borrow aliasing).
356            let instr = self.encoder_stream_pending[consumed..consumed + len].to_vec();
357            self.parse_encoder_instruction(&instr)?;
358            consumed += len;
359        }
360        self.encoder_stream_pending.drain(..consumed);
361
362        // Unblock every field section whose Required Insert Count has been
363        // reached, in arrival order.
364        let mut sections = Vec::new();
365        while let Some(front) = self.blocked.front() {
366            if front.ric > self.dynamic.inserted() {
367                break;
368            }
369            let front = self.blocked.pop_front().expect("front just inspected");
370            self.remove_blocked_section(front.stream_id);
371            let headers = self.decode_ready(&front.buf)?;
372            let size: usize = headers.iter().map(|(n, v)| n.len() + v.len()).sum();
373            if self.account_section(front.stream_id, size) > self.max_field_section_size {
374                return Err(QpackError::DecompressionFailed);
375            }
376            if front.ric > 0 {
377                self.acknowledge(front.ric);
378                self.emit_section_ack(front.stream_id);
379            }
380            sections.push(UnblockedSection {
381                stream_id: front.stream_id,
382                headers,
383            });
384        }
385
386        // Coalesced Insert Count Increment (2.2.2.3): the encoder may free
387        // references as soon as the received entries are acknowledged.
388        if self.dynamic.inserted() > self.known_received {
389            integer::encode(
390                &mut self.decoder_stream,
391                self.dynamic.inserted() - self.known_received,
392                6,
393                INSERT_COUNT_INCREMENT,
394            );
395            self.known_received = self.dynamic.inserted();
396        }
397        Ok(sections)
398    }
399
400    /// Parses a single *complete* encoder-stream instruction (RFC 9204
401    /// Section 4.3) from `instr` and materializes its dynamic-table update.
402    /// `feed_encoder_stream` guarantees `instr` holds a full instruction, so
403    /// the parses below cannot run out of bytes.
404    #[inline]
405    fn parse_encoder_instruction(&mut self, instr: &[u8]) -> Result<(), QpackError> {
406        let mut off = 0;
407        let header = instr[0];
408        off += 1;
409        match header & 0xC0 {
410            // `1 T` + 6-bit name index: Insert with Name Reference (4.3.2).
411            // The T bit being set masks to 0xC0, hence the two-arm pattern.
412            0x80 | 0xC0 => {
413                let index = integer::decode(instr, &mut off, 6, header).map_err(enc_stream_err)?;
414                let value = self
415                    .read_value_string(instr, &mut off)
416                    .map_err(enc_stream_err)?;
417                let name = if header & 0x40 != 0 {
418                    // T=1: static table.
419                    let idx = usize::try_from(index).map_err(|_| QpackError::EncoderStream)?;
420                    let (name, _) = static_table::get(idx).ok_or(QpackError::EncoderStream)?;
421                    Bytes::from_static(name)
422                } else {
423                    // T=0: dynamic table, relative index (index 0 is the most
424                    // recently inserted entry).
425                    let (name, _) = self
426                        .dynamic
427                        .get_relative_bytes(index)
428                        .ok_or(QpackError::EncoderStream)?;
429                    name
430                };
431                self.insert_entry(name, value)?;
432            }
433            0x40 => {
434                // Insert with Literal Name (4.3.3): the name length uses a
435                // 5-bit prefix, so `read_string` receives 6 (it reserves one
436                // bit for the Huffman flag).
437                let name = self
438                    .read_string(instr, &mut off, 6, header)
439                    .map_err(enc_stream_err)?;
440                let value = self
441                    .read_value_string(instr, &mut off)
442                    .map_err(enc_stream_err)?;
443                self.insert_entry(name, value)?;
444            }
445            _ => {
446                if header & 0x20 != 0 {
447                    // Set Dynamic Table Capacity (4.3.1).
448                    let capacity =
449                        integer::decode(instr, &mut off, 5, header).map_err(enc_stream_err)?;
450                    if capacity > self.max_capacity {
451                        return Err(QpackError::EncoderStream);
452                    }
453                    let evicted = self.dynamic.evict_for_capacity(capacity);
454                    if evicted > self.known_received {
455                        return Err(QpackError::EncoderStream);
456                    }
457                    self.dynamic.set_capacity(capacity);
458                } else {
459                    // Duplicate (4.3.4): relative index, 0 being the most
460                    // recently inserted entry.
461                    let index =
462                        integer::decode(instr, &mut off, 5, header).map_err(enc_stream_err)?;
463                    let (name, value) = self
464                        .dynamic
465                        .get_relative_bytes(index)
466                        .ok_or(QpackError::EncoderStream)?;
467                    self.insert_entry(name, value)?;
468                }
469            }
470        }
471        Ok(())
472    }
473
474    /// Decodes an encoded field section received on `stream_id`.
475    ///
476    /// Returns the decoded header list, or `None` when the section was
477    /// buffered as blocked (it is returned by a later
478    /// [`Decoder::feed_encoder_stream`] call).
479    ///
480    /// `now` is the caller's monotonic clock (any unit); it is recorded for
481    /// [`Decoder::expire_blocked`]. Sections that cannot be processed in
482    /// order are never decoded early: a section on a stream with buffered
483    /// blocked sections joins the queue even when it could be decoded
484    /// already (RFC 9204 Section 2.2.1 requires in-order processing).
485    #[inline]
486    pub fn decode_block(
487        &mut self,
488        buf: &[u8],
489        stream_id: u64,
490        now: u64,
491    ) -> Result<Option<Vec<(Bytes, Bytes)>>, QpackError> {
492        let (ric, _, _) = self.read_prefix(buf)?;
493        let stream_blocked = self.stream_has_blocked(stream_id);
494        if ric > self.dynamic.inserted() || stream_blocked {
495            if !stream_blocked && self.blocked_by_stream.len() >= self.max_blocked_streams {
496                return Err(QpackError::DecompressionFailed);
497            }
498            self.blocked.push_back(BlockedSection {
499                stream_id,
500                buf: Bytes::copy_from_slice(buf),
501                ric,
502                since: now,
503            });
504            self.add_blocked_section(stream_id);
505            return Ok(None);
506        }
507        let headers = self.decode_ready(buf)?;
508        let size: usize = headers.iter().map(|(n, v)| n.len() + v.len()).sum();
509        if self.account_section(stream_id, size) > self.max_field_section_size {
510            return Err(QpackError::DecompressionFailed);
511        }
512        if ric > 0 {
513            self.acknowledge(ric);
514            self.emit_section_ack(stream_id);
515        }
516        Ok(Some(headers))
517    }
518
519    /// Notifies that `stream_id` finished receiving (the peer closed its
520    /// send side): no further field sections can arrive on it, so its
521    /// field-section size budget is released. Safe to call when the stream
522    /// never carried a field section.
523    ///
524    /// Must only be called when no field section of the stream remains
525    /// buffered as blocked: a buffered section is decoded later by
526    /// [`Decoder::feed_encoder_stream`], which would then restart the
527    /// stream's budget from scratch. The HTTP/3 layer calls this only upon
528    /// observing the peer's stream end, which implies every section of the
529    /// stream (headers, trailers) has decoded.
530    #[inline]
531    pub fn stream_finished(&mut self, stream_id: u64) {
532        self.section_size_by_stream
533            .retain(|(id, _)| *id != stream_id);
534    }
535
536    /// Notifies that `stream_id` was reset or abandoned: buffered blocked
537    /// sections for it are dropped and a Stream Cancellation instruction is
538    /// queued (RFC 9204 Section 2.2.2.2). Returns the instruction.
539    #[inline]
540    pub fn stream_cancelled(&mut self, stream_id: u64) -> Bytes {
541        self.blocked.retain(|b| b.stream_id != stream_id);
542        self.blocked_by_stream.retain(|(id, _)| *id != stream_id);
543        self.section_size_by_stream
544            .retain(|(id, _)| *id != stream_id);
545        let mut out = Vec::new();
546        integer::encode(&mut out, stream_id, 6, STREAM_CANCELLATION);
547        self.decoder_stream.extend_from_slice(&out);
548        Bytes::from(out)
549    }
550
551    /// Drops blocked sections older than `max_age` in the caller's clock
552    /// units, queueing one Stream Cancellation per affected stream. Returns
553    /// the instructions.
554    #[inline]
555    pub fn expire_blocked(&mut self, now: u64, max_age: u64) -> Bytes {
556        let mut out = Vec::new();
557        let mut cancelled = Vec::new();
558        for blocked in &self.blocked {
559            if now.saturating_sub(blocked.since) > max_age
560                && !cancelled.contains(&blocked.stream_id)
561            {
562                cancelled.push(blocked.stream_id);
563                integer::encode(&mut out, blocked.stream_id, 6, STREAM_CANCELLATION);
564            }
565        }
566        if !cancelled.is_empty() {
567            // Cancelling a stream abandons every queued section on it, not
568            // merely the one whose timer fired. Keep the queue and its
569            // per-stream accounting in lockstep.
570            self.blocked
571                .retain(|blocked| !cancelled.contains(&blocked.stream_id));
572            self.blocked_by_stream
573                .retain(|(stream_id, _)| !cancelled.contains(stream_id));
574            self.section_size_by_stream
575                .retain(|(stream_id, _)| !cancelled.contains(stream_id));
576        }
577        self.decoder_stream.extend_from_slice(&out);
578        Bytes::from(out)
579    }
580
581    /// Decodes a field section whose Required Insert Count has been reached
582    /// (immediate or retried from the blocked queue), validating that the
583    /// announced Required Insert Count equals the largest referenced
584    /// absolute index plus one (RFC 9204 Section 2.1.2).
585    #[inline]
586    fn decode_ready(&self, buf: &[u8]) -> Result<Vec<(Bytes, Bytes)>, QpackError> {
587        let (ric, base, mut off) = self.read_prefix(buf)?;
588        let mut headers = Vec::new();
589        let mut needed = 0u64;
590        while off < buf.len() {
591            let header = buf[off];
592            off += 1;
593            if header & 0x80 != 0 {
594                // Indexed Field Line (4.5.2).
595                let index = integer::decode(buf, &mut off, 6, header).map_err(dec_failed)?;
596                if header & 0x40 != 0 {
597                    // Static table (T=1).
598                    let idx =
599                        usize::try_from(index).map_err(|_| QpackError::DecompressionFailed)?;
600                    let (name, value) =
601                        static_table::get(idx).ok_or(QpackError::DecompressionFailed)?;
602                    headers.push((Bytes::from_static(name), Bytes::from_static(value)));
603                } else {
604                    // Dynamic table (T=0): relative index from the Base.
605                    let (name, value) = self
606                        .dynamic
607                        .get_base_relative_bytes(base, index)
608                        .ok_or(QpackError::DecompressionFailed)?;
609                    needed = needed.max(base - index);
610                    headers.push((name, value));
611                }
612            } else if header & 0x40 != 0 {
613                // Literal Field Line with Name Reference (4.5.4).
614                let index = integer::decode(buf, &mut off, 4, header).map_err(dec_failed)?;
615                let value = self.read_value_string(buf, &mut off).map_err(dec_failed)?;
616                if header & 0x10 != 0 {
617                    // Static table (T=1).
618                    let idx =
619                        usize::try_from(index).map_err(|_| QpackError::DecompressionFailed)?;
620                    let (name, _) =
621                        static_table::get(idx).ok_or(QpackError::DecompressionFailed)?;
622                    headers.push((Bytes::from_static(name), value));
623                } else {
624                    // Dynamic table (T=0): relative index from the Base.
625                    let (name, _) = self
626                        .dynamic
627                        .get_base_relative_bytes(base, index)
628                        .ok_or(QpackError::DecompressionFailed)?;
629                    needed = needed.max(base - index);
630                    headers.push((name, value));
631                }
632            } else if header & 0x20 != 0 {
633                // Literal Field Line with Literal Name (4.5.6): the N bit is
634                // an instruction to peers not to index the line; it does not
635                // affect decoding.
636                let name = self
637                    .read_string(buf, &mut off, 4, header)
638                    .map_err(dec_failed)?;
639                let value = self.read_value_string(buf, &mut off).map_err(dec_failed)?;
640                headers.push((name, value));
641            } else if header & 0x10 != 0 {
642                // Indexed Field Line with Post-Base Index (4.5.3).
643                let index = integer::decode(buf, &mut off, 4, header).map_err(dec_failed)?;
644                let (name, value) = self
645                    .dynamic
646                    .get_post_base_bytes(base, index)
647                    .ok_or(QpackError::DecompressionFailed)?;
648                needed = needed.max(base + index + 1);
649                headers.push((name, value));
650            } else {
651                // Literal Field Line with Post-Base Name Reference (4.5.5).
652                let index = integer::decode(buf, &mut off, 3, header).map_err(dec_failed)?;
653                let (name, _) = self
654                    .dynamic
655                    .get_post_base_bytes(base, index)
656                    .ok_or(QpackError::DecompressionFailed)?;
657                needed = needed.max(base + index + 1);
658                let value = self.read_value_string(buf, &mut off).map_err(dec_failed)?;
659                headers.push((name, value));
660            }
661        }
662        if ric != needed {
663            return Err(QpackError::DecompressionFailed);
664        }
665        Ok(headers)
666    }
667
668    /// Parses the Encoded Field Section Prefix (RFC 9204 Section 4.5.1):
669    /// the Required Insert Count and the Base. Returns both and the number
670    /// of octets consumed.
671    #[inline]
672    fn read_prefix(&self, buf: &[u8]) -> Result<(u64, u64, usize), QpackError> {
673        let header = *buf.first().ok_or(QpackError::DecompressionFailed)?;
674        let mut off = 1;
675        let enc_ric = integer::decode(buf, &mut off, 8, header).map_err(dec_failed)?;
676        let max_entries = self.max_capacity / 32;
677        let ric = if enc_ric == 0 || max_entries == 0 {
678            0
679        } else {
680            let full_range = 2 * max_entries;
681            if enc_ric > full_range {
682                return Err(QpackError::DecompressionFailed);
683            }
684            let max_value = self.dynamic.inserted() + max_entries;
685            let max_wrapped = (max_value / full_range) * full_range;
686            let mut ric = max_wrapped + enc_ric - 1;
687            if ric > max_value {
688                if ric <= full_range {
689                    return Err(QpackError::DecompressionFailed);
690                }
691                ric -= full_range;
692            }
693            if ric == 0 {
694                return Err(QpackError::DecompressionFailed);
695            }
696            ric
697        };
698
699        // Base (4.5.1.2): Sign = 0 means Base = Ric + Delta; Sign = 1 means
700        // Base = Ric - Delta - 1.
701        let header = *buf.get(off).ok_or(QpackError::DecompressionFailed)?;
702        off += 1;
703        let delta = integer::decode(buf, &mut off, 7, header).map_err(dec_failed)?;
704        let base = if header & 0x80 != 0 {
705            ric.checked_sub(delta + 1)
706                .ok_or(QpackError::DecompressionFailed)?
707        } else {
708            ric.checked_add(delta)
709                .ok_or(QpackError::DecompressionFailed)?
710        };
711        Ok((ric, base, off))
712    }
713
714    /// Reads an N-bit-prefix string literal (RFC 9204 Section 4.1.2) at
715    /// `off`, advancing `off` past it. `header` is an already-consumed octet
716    /// carrying the Huffman bit (its `prefix_bits - 1` bit) and the length
717    /// prefix; for name strings it is the field line's or instruction's
718    /// first octet, which pairs with the name length prefix.
719    #[inline]
720    fn read_string(
721        &self,
722        buf: &[u8],
723        off: &mut usize,
724        prefix_bits: u8,
725        header: u8,
726    ) -> Result<Bytes, HpackError> {
727        let huffman = header & (1 << (prefix_bits - 1)) != 0;
728        let len = integer::decode(buf, off, prefix_bits - 1, header)?;
729        let len = usize::try_from(len).map_err(|_| HpackError::InvalidString)?;
730        let end = (*off).checked_add(len).ok_or(HpackError::InvalidString)?;
731        let src = buf.get(*off..end).ok_or(HpackError::InvalidString)?;
732        *off = end;
733        if huffman {
734            let mut dst = Vec::with_capacity(len);
735            huffman::decode(src, &mut dst)?;
736            Ok(Bytes::from(dst))
737        } else {
738            Ok(Bytes::copy_from_slice(src))
739        }
740    }
741
742    /// Reads an 8-bit-prefix string literal whose length octet is next in
743    /// the buffer, advancing `off` past it. This is the form of every value
744    /// string (RFC 9204 Section 4.5).
745    #[inline]
746    fn read_value_string(&self, buf: &[u8], off: &mut usize) -> Result<Bytes, HpackError> {
747        let header = *buf.get(*off).ok_or(HpackError::InvalidString)?;
748        *off += 1;
749        self.read_string(buf, off, 8, header)
750    }
751
752    /// Inserts an entry named `name` with `value`, enforcing the eviction
753    /// rules of RFC 9204 Sections 2.1.1 and 3.2.2: entries with an absolute
754    /// index at or above the Known Received Count are not evictable, so an
755    /// insert that would evict them is an encoder error.
756    #[inline]
757    fn insert_entry(&mut self, name: Bytes, value: Bytes) -> Result<(), QpackError> {
758        let size = DynamicTable::entry_size(&name, &value);
759        let evicted = self.dynamic.would_evict(size);
760        if evicted > self.known_received {
761            return Err(QpackError::EncoderStream);
762        }
763        self.dynamic
764            .insert(name, value)
765            .map_err(|_| QpackError::EncoderStream)
766    }
767
768    /// Raises the Known Received Count to `ric` (RFC 9204 Section 2.1.4).
769    #[inline]
770    fn acknowledge(&mut self, ric: u64) {
771        self.known_received = self.known_received.max(ric);
772    }
773
774    /// Queues a Section Acknowledgment for `stream_id`, unless the maximum
775    /// dynamic table capacity is zero (the encoder has no dynamic table
776    /// references to free).
777    #[inline]
778    fn emit_section_ack(&mut self, stream_id: u64) {
779        integer::encode(&mut self.decoder_stream, stream_id, 7, SECTION_ACK);
780    }
781
782    /// Whether `stream_id` has buffered blocked sections.
783    #[inline]
784    fn stream_has_blocked(&self, stream_id: u64) -> bool {
785        self.blocked_by_stream
786            .iter()
787            .any(|(id, _)| *id == stream_id)
788    }
789
790    /// Records a newly buffered field section without rescanning the queue.
791    #[inline]
792    fn add_blocked_section(&mut self, stream_id: u64) {
793        if let Some((_, count)) = self
794            .blocked_by_stream
795            .iter_mut()
796            .find(|(id, _)| *id == stream_id)
797        {
798            *count += 1;
799        } else {
800            self.blocked_by_stream.push((stream_id, 1));
801        }
802    }
803
804    /// Charges `size` against `stream_id`'s field-section budget and
805    /// returns the stream's new cumulative total. Creates the entry on
806    /// first use.
807    ///
808    /// `size` is the decoded name and value octets, not the size of the
809    /// encoded block: Huffman encoding and index references make the two
810    /// diverge arbitrarily (RFC 9114 Section 7.2.4.1).
811    #[inline]
812    fn account_section(&mut self, stream_id: u64, size: usize) -> usize {
813        if let Some((_, total)) = self
814            .section_size_by_stream
815            .iter_mut()
816            .find(|(id, _)| *id == stream_id)
817        {
818            *total = total.saturating_add(size);
819            *total
820        } else {
821            self.section_size_by_stream.push((stream_id, size));
822            size
823        }
824    }
825
826    /// Removes one decoded blocked section from the per-stream count.
827    #[inline]
828    fn remove_blocked_section(&mut self, stream_id: u64) {
829        let Some(index) = self
830            .blocked_by_stream
831            .iter()
832            .position(|(id, _)| *id == stream_id)
833        else {
834            debug_assert!(false, "blocked section missing stream accounting");
835            return;
836        };
837        if self.blocked_by_stream[index].1 == 1 {
838            self.blocked_by_stream.swap_remove(index);
839        } else {
840            self.blocked_by_stream[index].1 -= 1;
841        }
842    }
843}
844
845/// Maps an `HpackError` from an encoder stream instruction to
846/// `QPACK_ENCODER_STREAM_ERROR`.
847#[inline]
848fn enc_stream_err(_: HpackError) -> QpackError {
849    QpackError::EncoderStream
850}
851
852/// Maps an `HpackError` from a field section to `QPACK_DECOMPRESSION_FAILED`.
853#[inline]
854fn dec_failed(_: HpackError) -> QpackError {
855    QpackError::DecompressionFailed
856}
857
858/// Maps an `HpackError` from a decoder stream instruction to
859/// `QPACK_DECODER_STREAM_ERROR`.
860#[inline]
861fn dec_stream_err(_: HpackError) -> QpackError {
862    QpackError::DecoderStream
863}
864#[cfg(test)]
865mod tests;