structured_zstd/decoding/frame_decoder.rs
1//! Framedecoder is the main low-level struct users interact with to decode zstd frames
2//!
3//! Zstandard compressed data is made of one or more frames. Each frame is independent and can be
4//! decompressed independently of other frames. This module contains structures
5//! and utilities that can be used to decode a frame.
6
7use super::frame;
8use crate::decoding;
9use crate::decoding::block_decoder::BlockDecoder;
10use crate::decoding::buffer_backend::BufferBackend;
11use crate::decoding::dictionary::{Dictionary, DictionaryHandle};
12use crate::decoding::errors::{DecodeBlockContentError, FrameDecoderError};
13use crate::decoding::flat_buf::FlatBuf;
14use crate::decoding::ringbuffer::RingBuffer;
15use crate::decoding::scratch::DecoderScratch;
16use crate::io::{Error, Read, Write};
17use alloc::collections::BTreeMap;
18use alloc::vec::Vec;
19use core::convert::TryInto;
20
21use crate::common::MAXIMUM_ALLOWED_WINDOW_SIZE;
22
23/// Build the block-header decode error. With the `lsm` feature it captures
24/// the failing block's index and frame offset (block-precise recovery);
25/// without it, the legacy positionless variant — so the default build's
26/// error surface stays byte-identical to the upstream zstd.
27#[cfg(feature = "lsm")]
28fn block_header_decode_error(
29 source: crate::decoding::errors::BlockHeaderReadError,
30 block_index: u32,
31 frame_offset: u32,
32) -> FrameDecoderError {
33 FrameDecoderError::FailedToReadBlockHeaderAt {
34 source,
35 block_index,
36 frame_offset,
37 }
38}
39#[cfg(not(feature = "lsm"))]
40fn block_header_decode_error(
41 source: crate::decoding::errors::BlockHeaderReadError,
42 _block_index: u32,
43 _frame_offset: u32,
44) -> FrameDecoderError {
45 FrameDecoderError::FailedToReadBlockHeader(source)
46}
47
48/// Build the block-body decode error. With `lsm` it captures the block
49/// index, frame offset, and the failing block's structural metadata
50/// (reconstructed from its header); without it, the legacy variant.
51#[cfg(feature = "lsm")]
52fn block_body_decode_error(
53 source: DecodeBlockContentError,
54 block_index: u32,
55 frame_offset: u32,
56 header: &crate::blocks::block::BlockHeader,
57 header_size: u8,
58) -> FrameDecoderError {
59 use crate::blocks::block::BlockType;
60 // Physical wire body vs the raw `Block_Size` field: RLE writes a single
61 // body byte while `Block_Size` carries the repeat count; Raw/Compressed
62 // bodies match the field.
63 let (body_size, block_size_field) = match header.block_type {
64 BlockType::RLE => (1u32, header.decompressed_size),
65 _ => (header.content_size, header.content_size),
66 };
67 FrameDecoderError::FailedToReadBlockBodyAt {
68 source,
69 block_index,
70 frame_offset,
71 block: crate::encoding::frame_emit_info::FrameBlock {
72 offset_in_frame: frame_offset,
73 header_size,
74 body_size,
75 block_size_field,
76 block_type: header.block_type,
77 last_block: header.last_block,
78 // Raw/RLE carry their regenerated size in the header;
79 // a Compressed block's is unknown until decoded, so
80 // `read_block_header` leaves `decompressed_size` 0 here.
81 decompressed_size: header.decompressed_size,
82 },
83 }
84}
85#[cfg(not(feature = "lsm"))]
86fn block_body_decode_error(
87 source: DecodeBlockContentError,
88 _block_index: u32,
89 _frame_offset: u32,
90 _header: &crate::blocks::block::BlockHeader,
91 _header_size: u8,
92) -> FrameDecoderError {
93 FrameDecoderError::FailedToReadBlockBody(source)
94}
95
96/// Low level Zstandard decoder that can be used to decompress frames with fine control over when and how many bytes are decoded.
97///
98/// This decoder is able to decode frames only partially and gives control
99/// over how many bytes/blocks will be decoded at a time (so you don't have to decode a 10GB file into memory all at once).
100/// It reads bytes as needed from a provided source and can be read from to collect partial results.
101///
102/// If you want to just read the whole frame with an `io::Read` without having to deal with manually calling [FrameDecoder::decode_blocks]
103/// you can use the provided [crate::decoding::StreamingDecoder] wich wraps this FrameDecoder.
104///
105/// Workflow is as follows:
106/// ```
107/// use structured_zstd::decoding::BlockDecodingStrategy;
108///
109/// # #[cfg(feature = "std")]
110/// use std::io::{Read, Write};
111///
112/// // no_std environments can use the crate's own Read traits
113/// # #[cfg(not(feature = "std"))]
114/// use structured_zstd::io::{Read, Write};
115///
116/// fn decode_this(mut file: impl Read) {
117/// //Create a new decoder
118/// let mut frame_dec = structured_zstd::decoding::FrameDecoder::new();
119/// let mut result = Vec::new();
120///
121/// // Use reset or init to make the decoder ready to decode the frame from the io::Read
122/// frame_dec.reset(&mut file).unwrap();
123///
124/// // Loop until the frame has been decoded completely
125/// while !frame_dec.is_finished() {
126/// // decode (roughly) batch_size many bytes
127/// frame_dec.decode_blocks(&mut file, BlockDecodingStrategy::UptoBytes(1024)).unwrap();
128///
129/// // read from the decoder to collect bytes from the internal buffer
130/// let bytes_read = frame_dec.read(result.as_mut_slice()).unwrap();
131///
132/// // then do something with it
133/// do_something(&result[0..bytes_read]);
134/// }
135///
136/// // handle the last chunk of data
137/// while frame_dec.can_collect() > 0 {
138/// let x = frame_dec.read(result.as_mut_slice()).unwrap();
139///
140/// do_something(&result[0..x]);
141/// }
142/// }
143///
144/// fn do_something(data: &[u8]) {
145/// # #[cfg(feature = "std")]
146/// std::io::stdout().write_all(data).unwrap();
147/// }
148/// ```
149pub struct FrameDecoder {
150 state: Option<FrameDecoderState>,
151 /// Test-only observability: frames decoded via `run_direct_decode`.
152 /// The direct and buffered paths are byte-identical, so dispatch
153 /// regressions (e.g. re-excluding dictionary frames from the direct
154 /// gate) are invisible to output assertions; tests pin the path here.
155 #[cfg(test)]
156 direct_frames: u64,
157 // Registered dictionaries are stored by shared handle (Arc/Rc) so a
158 // single content copy is referenced by every frame the decoder decodes
159 // (upstream zstd `ZSTD_refDDict`), rather than re-copied into the decode buffer
160 // per frame. `add_dict` wraps an owned `Dictionary` into a handle.
161 owned_dicts: BTreeMap<u32, DictionaryHandle>,
162 #[cfg(target_has_atomic = "ptr")]
163 shared_dicts: BTreeMap<u32, DictionaryHandle>,
164 #[cfg(not(target_has_atomic = "ptr"))]
165 shared_dicts: (),
166 /// `ZSTD_f_zstd1_magicless` — when true, [`init`] / [`reset`]
167 /// expect frames without the 4-byte magic number prefix.
168 /// Default false (standard zstd format).
169 magicless: bool,
170 /// How the optional content checksum is handled. Default
171 /// [`ContentChecksum::EmitOnly`] (compute + expose, no error on
172 /// mismatch). Set via [`Self::set_content_checksum`].
173 content_checksum: ContentChecksum,
174 /// Pinned `Dictionary_ID` expectation set via
175 /// [`Self::expect_dict_id`]. `None` (default) disables the
176 /// check; `Some(0)` matches frames whose header omits the
177 /// optional dict_id (treated as "no dictionary"). Validated in
178 /// [`Self::reset`] AFTER the frame header parses successfully
179 /// and BEFORE any block decode work.
180 #[cfg(feature = "lsm")]
181 expect_dict_id: Option<u32>,
182 /// Pinned `Window_Descriptor` byte expectation set via
183 /// [`Self::expect_window_descriptor`]. `None` (default)
184 /// disables the check. Validated in [`Self::reset`] AFTER the
185 /// frame header parses successfully and BEFORE any block
186 /// decode work. Single-segment frames (which omit the
187 /// `Window_Descriptor` byte from the wire) surface as
188 /// [`crate::decoding::errors::FrameDecoderError::UnexpectedWindowDescriptor`]
189 /// with `found: None`.
190 #[cfg(feature = "lsm")]
191 expect_window_descriptor: Option<u8>,
192 /// When `true`, the per-block decode loop XXH64-hashes each
193 /// block's decompressed bytes and stores the low-32-bit digest in
194 /// [`Self::computed_block_checksums`]. Default `false` (zero
195 /// cost). Set via [`Self::enable_per_block_checksums`]. Gated on
196 /// `all(lsm, hash)` because XXH64 lives behind the `hash`
197 /// feature.
198 #[cfg(all(feature = "lsm", feature = "hash"))]
199 per_block_checksums_enabled: bool,
200 /// Per-block XXH64 (low 32 bits) digests captured during the
201 /// current frame's decode when `per_block_checksums_enabled` is
202 /// set. Reset at the start of every new frame. Gated on
203 /// `all(lsm, hash)` (see `per_block_checksums_enabled`).
204 #[cfg(all(feature = "lsm", feature = "hash"))]
205 computed_block_checksums: alloc::vec::Vec<u32>,
206 /// Best kernel this CPU offers, resolved once per decoder rather than per
207 /// block decoder built. A chunked decode enters `decode_from_to` once per
208 /// caller-sized target, so detecting there put an atomic read and a branch
209 /// on every call; feature detection belongs before the work, not inside it.
210 kernel: crate::cpu_kernel::CpuKernelTag,
211}
212
213/// How the decoder treats a frame's optional XXH64 content checksum
214/// (RFC 8878 Content_Checksum_flag). The XXH64 pass over the decompressed
215/// output is a measurable share of decode time, so it is made skippable.
216///
217/// ```
218/// use structured_zstd::decoding::{ContentChecksum, FrameDecoder};
219/// let mut decoder = FrameDecoder::new();
220/// decoder.set_content_checksum(ContentChecksum::Verify);
221/// ```
222#[derive(Copy, Clone, PartialEq, Eq, Debug, Default)]
223pub enum ContentChecksum {
224 /// Skip the XXH64 pass entirely: no compute, no verify.
225 /// `get_calculated_checksum()` returns `None`.
226 None,
227 /// Compute the checksum and expose it via the accessors, but do not
228 /// error on a mismatch. This is the default and matches the historical
229 /// behaviour (callers verify manually if they wish).
230 #[default]
231 EmitOnly,
232 /// Compute the checksum and compare it against the frame's stored value;
233 /// a disagreement fails the decode with
234 /// [`FrameDecoderError::ChecksumMismatch`](crate::decoding::errors::FrameDecoderError::ChecksumMismatch).
235 /// Without the `hash` feature there is no way to compute a digest, so
236 /// `Verify` cannot detect a mismatch and behaves like `None`.
237 Verify,
238}
239
240/// Decode-relevant identity of a frame, used to reject a [`ResumeState`]
241/// captured from one frame being applied to a frame of a different shape. Covers
242/// every header field that changes how blocks decode (buffer sizing, backend
243/// kind, entropy/dictionary context, trailing-checksum handling, declared
244/// content size, magicless framing).
245///
246/// This is a SHAPE guard, not a content-unique fingerprint: two distinct frames
247/// that happen to share all these header fields produce the same key (no cheap
248/// header field uniquely identifies frame content). It catches the realistic
249/// accidental misuse — applying a snapshot to a frame with a different
250/// window/dictionary/size — with a typed error instead of byte-wrong output.
251/// Pairing a `ResumeState` with the correct frame's compressed source and
252/// `window_prime` remains the caller's contract.
253#[cfg(feature = "lsm")]
254#[derive(Clone, Copy, PartialEq, Eq, Debug)]
255struct FrameKey {
256 window_size: u64,
257 frame_content_size: u64,
258 /// `Dictionary_ID` declared in the frame header (`None` when omitted).
259 dictionary_id: Option<u32>,
260 /// Dictionary actually applied to the decoder (`state.using_dict`). This is
261 /// distinct from `dictionary_id`: a frame with a dictless header can still
262 /// be decoded with an explicit dictionary via `reset_with_dict_handle` /
263 /// `force_dict`, and two such decodes with different dictionaries must NOT
264 /// compare equal — keying only on the header field would miss that.
265 active_dictionary_id: Option<u32>,
266 single_segment: bool,
267 content_checksum: bool,
268 magicless: bool,
269}
270
271#[cfg(feature = "lsm")]
272impl FrameKey {
273 fn from_state(state: &FrameDecoderState, magicless: bool) -> FrameKey {
274 let header = &state.frame_header;
275 FrameKey {
276 window_size: header.window_size().unwrap_or(0),
277 frame_content_size: header.frame_content_size(),
278 dictionary_id: header.dictionary_id(),
279 active_dictionary_id: state.using_dict,
280 single_segment: header.descriptor.single_segment_flag(),
281 content_checksum: header.descriptor.content_checksum_flag(),
282 magicless,
283 }
284 }
285}
286
287/// XXH64 of a contiguous byte slice — the resume-side counterpart to
288/// [`DecoderScratchKind::window_tail_hash`]. Streaming XXH64 is chunk-boundary
289/// independent, so this single-slice hash equals the emit-side two-slice hash
290/// over the same bytes.
291#[cfg(all(feature = "lsm", feature = "hash"))]
292fn xxh64_of(bytes: &[u8]) -> u64 {
293 use core::hash::Hasher;
294 let mut h = twox_hash::XxHash64::with_seed(0);
295 h.write(bytes);
296 h.finish()
297}
298
299/// Cross-block decode state needed to resume a cold partial decode at an inner
300/// block boundary, emitted by [`FrameDecoder::decode_blocks_partial`] when its
301/// `emit_resume` argument is `true` (returned in
302/// [`PartialDecode::resume_state`]) and fed back via that same method's
303/// [`resume`](FrameDecoder::decode_blocks_partial) argument
304/// ([`ResumeInput`]).
305///
306/// A zstd block does not carry all the state required to decode it in
307/// isolation: besides the shared match window (the decompressed output history),
308/// a Compressed block may reuse the previous block's entropy tables via
309/// `Repeat_Mode` (literals Huffman + the LL/OF/ML FSE distributions) and always
310/// continues the running repeat-offset history. This snapshot carries exactly
311/// that carry-over state plus the resume coordinates, so resuming is
312/// byte-identical to a contiguous decode even across a dropped decoder. The
313/// window itself is NOT stored here — the caller supplies it back through
314/// [`ResumeInput::window_prime`] from the decompressed output it already
315/// persists. Neither is the dictionary: for a dictionary frame the caller
316/// re-attaches it to the resuming decoder via [`FrameDecoder::reset`] /
317/// [`FrameDecoder::reset_with_dict_handle`] (it already holds the dictionary
318/// from encode time), and the snapshot records only the dictionary's identity
319/// so a resume under a different dictionary is rejected.
320///
321/// Behind the `lsm` Cargo feature.
322#[cfg(feature = "lsm")]
323#[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
324pub struct ResumeState {
325 /// Identity of the frame this state was captured from. Compared against the
326 /// frame currently reset into the decoder before any state is restored, so a
327 /// snapshot from a different frame shape is rejected with
328 /// [`FrameDecoderError::ResumeFrameMismatch`] instead of silently producing
329 /// byte-wrong output.
330 frame_key: FrameKey,
331 /// Index of the block to resume AT (the first block NOT yet decoded).
332 block_index: u32,
333 /// Cumulative decompressed byte count produced before `block_index`.
334 output_offset: u64,
335 /// FSE tables (LL/OF/ML) as of the last decoded block — the source for a
336 /// `Repeat_Mode` resume block.
337 fse: crate::decoding::scratch::FSEScratch,
338 /// Huffman literals table as of the last decoded block — the source for a
339 /// treeless (repeat) literals resume block.
340 huf: crate::decoding::scratch::HuffmanScratch,
341 /// Running repeat-offset history (`offset_hist`) as of the last decoded
342 /// block.
343 offset_hist: [u32; 3],
344 /// XXH64 of the exact window-prime bytes (the last `min(window_size,
345 /// output_offset)` decompressed bytes) captured at emit. Verified at resume
346 /// against the caller-supplied [`ResumeInput::window_prime`]: a content
347 /// mismatch (wrong frame, wrong or corrupted prime) is a near-unique
348 /// (≈2⁻⁶⁴) signal and is rejected with
349 /// [`FrameDecoderError::ResumeFrameMismatch`]. This is the content-exact
350 /// guard; [`FrameKey`] is the cheap shape pre-check that works without the
351 /// `hash` feature. Behind `all(lsm, hash)`.
352 #[cfg(feature = "hash")]
353 window_hash: u64,
354}
355
356#[cfg(feature = "lsm")]
357impl ResumeState {
358 /// Inner block index this state resumes at (the first block not yet
359 /// decoded). Pass it as the `end_block` lower bound (and as `start_block`)
360 /// of the resuming
361 /// [`decode_blocks_partial`](FrameDecoder::decode_blocks_partial) call.
362 pub fn block_index(&self) -> u32 {
363 self.block_index
364 }
365
366 /// Cumulative decompressed byte count produced before
367 /// [`block_index`](Self::block_index) — i.e. the decompressed offset at
368 /// which the resumed output begins. Equals
369 /// `FrameEmitInfo::decompressed_byte_range(block_index).start`. Use it to
370 /// slice the `window_prime` tail the resumed call needs.
371 pub fn output_offset(&self) -> u64 {
372 self.output_offset
373 }
374}
375
376// Manual Debug: the entropy tables are large internal scratch with no useful
377// Debug surface; only the resume coordinates are worth printing (and this lets
378// `PartialDecode` keep its derived Debug).
379#[cfg(feature = "lsm")]
380impl core::fmt::Debug for ResumeState {
381 fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
382 f.debug_struct("ResumeState")
383 .field("block_index", &self.block_index)
384 .field("output_offset", &self.output_offset)
385 .finish_non_exhaustive()
386 }
387}
388
389/// Resume input fed to [`FrameDecoder::decode_blocks_partial`]'s `resume`
390/// argument to continue a cold partial decode without re-decompressing the
391/// preceding blocks.
392///
393/// Behind the `lsm` Cargo feature.
394#[cfg(feature = "lsm")]
395#[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
396pub struct ResumeInput<'a> {
397 /// The caller's already-decompressed output ending just before
398 /// [`ResumeState::block_index`]. Must contain at least the last
399 /// `min(window_size, output_offset)` bytes (a full match window, or the
400 /// whole prefix when it is shorter than one window); anything beyond the
401 /// last `window_size` bytes is ignored, so passing the entire prefix is
402 /// also valid (capped internally, bounding resume memory to one window).
403 pub window_prime: &'a [u8],
404 /// Cross-block entropy/repcode state emitted by the prior
405 /// [`decode_blocks_partial`](FrameDecoder::decode_blocks_partial) call.
406 pub state: &'a ResumeState,
407}
408
409/// Backend-tagged decode scratch — chosen at frame-reset time based
410/// on the parsed `FrameHeader.descriptor.single_segment_flag()` and
411/// kept stable through the lifetime of the frame. The match in each
412/// helper below dispatches **once per call** (e.g. once per block in
413/// `decode_block_content`, once per drain in `drain_to_writer`) —
414/// never inside the hot push/repeat loop, which is fully
415/// monomorphised through the `DecoderScratch<B>` generic.
416enum DecoderScratchKind {
417 Ring(DecoderScratch<RingBuffer>),
418 Flat(DecoderScratch<FlatBuf>),
419}
420
421impl DecoderScratchKind {
422 fn new_ring(window_size: usize) -> Self {
423 // Lazy ring-buffer allocation: do NOT `reserve(window_size)` here.
424 // The direct-decode path (`run_direct_decode`) writes through
425 // `UserSliceBackend` and never touches the ring; allocating it
426 // eagerly wastes one full window of peak memory on the common
427 // direct-eligible frame. On the non-direct path the window is
428 // pre-reserved once at frame entry (`decode_all_impl` and
429 // `decode_blocks` both call `DecoderScratchKind::reserve_buffer`
430 // before any block writes), so multi-block frames pay one
431 // amortised grow instead of repeated `reserve_amortized` steps
432 // per block. Issue #279 round 2.
433 let s = DecoderScratch::<RingBuffer>::new(window_size);
434 Self::Ring(s)
435 }
436
437 /// Construct a flat-backed scratch for a single-segment frame.
438 /// `frame_content_size` is the upcoming output size in bytes
439 /// (== `window_size` when the flag is set).
440 ///
441 /// Lazy buffer allocation (mirrors [`Self::new_ring`]): do NOT
442 /// pre-size the `FlatBuf`. The direct-decode path
443 /// (`run_direct_decode`) writes through `UserSliceBackend` and never
444 /// touches this buffer, so eagerly allocating a full FCS wastes one
445 /// whole content-size of peak memory on the common direct-eligible
446 /// single-segment frame. The non-direct fallback reserves it once via
447 /// `reserve_buffer(window_size)` at frame entry before any block
448 /// write (`FlatBuf::reserve` adds the `WILDCOPY_OVERLENGTH` slack),
449 /// and every inline-exec site (trait method and per-kernel macros)
450 /// now carries a tight-tail bounded copy, so a tight buffer can never
451 /// overshoot regardless of construction-time slack.
452 fn new_flat(frame_content_size: usize) -> Self {
453 let s = DecoderScratch::<FlatBuf>::new(frame_content_size);
454 Self::Flat(s)
455 }
456
457 /// Reset (or transition between) backends for a new frame.
458 /// Reuses the existing `DecoderScratch` allocations (FSE / HUF
459 /// tables, sequence vec, etc.) when the backend kind is unchanged
460 /// — only the underlying buffer is re-sized for the new frame.
461 /// Building a fresh `DecoderScratch` on every frame would
462 /// re-allocate everything and was measured at +255 % vs ring on
463 /// small frames; reusing it keeps the small-frame cost flat.
464 fn reset(&mut self, frame: &frame::FrameHeader, window_size: usize) {
465 if frame.descriptor.single_segment_flag() {
466 match self {
467 Self::Flat(s) => {
468 s.reset(window_size);
469 // `DecoderScratch::reset` clears the backing buffer and
470 // updates `window_size` WITHOUT reserving it (it may still
471 // resize the per-block scratch Vecs up to
472 // `min(window_size, MAX_BLOCK_SIZE)`). Backing-buffer
473 // capacity is decided one layer up: direct-eligible frames
474 // never touch it, and the non-direct path pre-reserves once
475 // via `reserve_buffer(window_size)` at frame entry.
476 }
477 Self::Ring(_) => *self = Self::new_flat(window_size),
478 }
479 } else {
480 match self {
481 Self::Ring(s) => s.reset(window_size),
482 Self::Flat(_) => *self = Self::new_ring(window_size),
483 }
484 }
485 }
486
487 fn init_from_dict(&mut self, dict: &DictionaryHandle) {
488 match self {
489 Self::Ring(s) => s.init_from_dict(dict),
490 Self::Flat(s) => s.init_from_dict(dict),
491 }
492 }
493
494 #[inline]
495 fn buffer_len(&self) -> usize {
496 match self {
497 Self::Ring(s) => s.buffer.len(),
498 Self::Flat(s) => s.buffer.len(),
499 }
500 }
501
502 fn workspace_bytes(&self) -> usize {
503 match self {
504 Self::Ring(s) => s.workspace_bytes(),
505 Self::Flat(s) => s.workspace_bytes(),
506 }
507 }
508
509 /// Pre-reserve the backing buffer to `window_size` in a single
510 /// allocation. Called once on the non-direct (`decode_blocks`) path
511 /// after direct-eligibility is ruled out, so multi-segment fallback
512 /// decodes don't pay repeated `reserve_amortized` grow steps
513 /// (128 KiB → 256 KiB → ... → window) as blocks accumulate.
514 ///
515 /// Direct-eligible frames never call this and pay zero backing-buffer
516 /// allocation for the window, on BOTH backends: `new_ring` and
517 /// `new_flat` are each lazy (no pre-reserve), so a direct-eligible
518 /// frame writes only through `UserSliceBackend` and leaves this
519 /// buffer empty.
520 ///
521 /// `target` is the TARGET buffer capacity (`decoding_buffer_size`) and
522 /// `growth_limit` the most later growth may reach
523 /// (`decoding_buffer_limit`); the method itself computes the shortfall
524 /// past the bytes already buffered before calling the backend's
525 /// ADDITIONAL-semantics `reserve_exact`. That keeps re-entries (the
526 /// decode_all fallback loop runs `decode_blocks` once per strategy
527 /// chunk, and streaming callers invoke it per call) from growing a
528 /// window-full buffer toward 2x window, while per-block growth keeps
529 /// the amortized `reserve`.
530 /// Hand the buffer what the frame declared it would produce, so the
531 /// per-block reservation can stop at the frame's remainder.
532 #[inline]
533 fn set_declared_content(&mut self, content_size: Option<u64>) {
534 match self {
535 Self::Ring(s) => s.buffer.set_declared_content(content_size),
536 Self::Flat(s) => s.buffer.set_declared_content(content_size),
537 }
538 }
539
540 #[inline]
541 fn reserve_buffer(&mut self, target: usize, growth_limit: usize) {
542 let window_size = target;
543 // Exact growth: this is the one-shot pre-reservation, and a request
544 // landing one slack past the retained capacity (e.g. a dictionary
545 // prefix already loaded into the buffer) must not DOUBLE a
546 // window-sized allocation through the amortized policy. Per-block
547 // growth keeps the amortized `reserve`.
548 //
549 // `reserve_exact` takes ADDITIONAL capacity, so request only the
550 // shortfall past the bytes already buffered: the decode_all
551 // fallback loop re-enters `decode_blocks` once per strategy chunk,
552 // and re-requesting the full window each iteration would grow a
553 // window-sized buffer toward 2x window.
554 match self {
555 Self::Ring(s) => {
556 // The ring's growth rounds to the next power of two, capped at
557 // its limit; with the limit at the target for this one-shot
558 // reservation it lands exactly there, and later growth (a
559 // compressed block's own reservation) may then reach the
560 // frame's real limit, never double past it.
561 s.buffer.set_growth_limit(window_size);
562 let additional = window_size.saturating_sub(s.buffer.len());
563 s.buffer.reserve_exact(additional);
564 s.buffer.set_growth_limit(growth_limit);
565 }
566 Self::Flat(s) => {
567 let additional = window_size.saturating_sub(s.buffer.len());
568 s.buffer.reserve_exact(additional);
569 }
570 }
571 }
572
573 /// Last `n` bytes of the visible buffer as `(s1, s2)` (wrap-aware).
574 /// Routes through whichever backend the current scratch holds.
575 #[cfg(all(feature = "lsm", feature = "hash"))]
576 fn last_n_as_slices(&self, n: usize) -> (&[u8], &[u8]) {
577 match self {
578 Self::Ring(s) => s.buffer.last_n_as_slices(n),
579 Self::Flat(s) => s.buffer.last_n_as_slices(n),
580 }
581 }
582
583 fn buffer_drain(&mut self) -> Vec<u8> {
584 match self {
585 Self::Ring(s) => s.buffer.drain(),
586 Self::Flat(s) => s.buffer.drain(),
587 }
588 }
589
590 fn buffer_drain_to_window_size(&mut self) -> Option<Vec<u8>> {
591 match self {
592 Self::Ring(s) => s.buffer.drain_to_window_size(),
593 Self::Flat(s) => s.buffer.drain_to_window_size(),
594 }
595 }
596
597 fn buffer_drain_to_writer(&mut self, sink: impl Write) -> Result<usize, Error> {
598 match self {
599 Self::Ring(s) => s.buffer.drain_to_writer(sink),
600 Self::Flat(s) => s.buffer.drain_to_writer(sink),
601 }
602 }
603
604 fn buffer_drain_to_window_size_writer(&mut self, sink: impl Write) -> Result<usize, Error> {
605 match self {
606 Self::Ring(s) => s.buffer.drain_to_window_size_writer(sink),
607 Self::Flat(s) => s.buffer.drain_to_window_size_writer(sink),
608 }
609 }
610
611 fn buffer_can_drain(&self) -> usize {
612 match self {
613 Self::Ring(s) => s.buffer.can_drain(),
614 Self::Flat(s) => s.buffer.can_drain(),
615 }
616 }
617
618 fn buffer_can_drain_to_window_size(&self) -> Option<usize> {
619 match self {
620 Self::Ring(s) => s.buffer.can_drain_to_window_size(),
621 Self::Flat(s) => s.buffer.can_drain_to_window_size(),
622 }
623 }
624
625 fn buffer_read(&mut self, target: &mut [u8]) -> Result<usize, Error> {
626 match self {
627 Self::Ring(s) => s.buffer.read(target),
628 Self::Flat(s) => s.buffer.read(target),
629 }
630 }
631
632 /// [`Self::buffer_read`], also reporting the drainable bytes `target` had
633 /// no room for.
634 fn buffer_read_reporting_pending(
635 &mut self,
636 target: &mut [u8],
637 ) -> Result<(usize, usize), Error> {
638 match self {
639 Self::Ring(s) => s.buffer.read_reporting_pending(target),
640 Self::Flat(s) => s.buffer.read_reporting_pending(target),
641 }
642 }
643
644 fn buffer_read_all(&mut self, target: &mut [u8]) -> Result<usize, Error> {
645 match self {
646 Self::Ring(s) => s.buffer.read_all(target),
647 Self::Flat(s) => s.buffer.read_all(target),
648 }
649 }
650
651 /// Drop visible output beyond `window_size` without producing it,
652 /// keeping the most recent `window_size` bytes available to back
653 /// future match copies. Used by `decode_blocks_partial` to bound
654 /// memory while decoding the leading (skipped) blocks into the window.
655 #[cfg(feature = "lsm")]
656 fn buffer_drop_to_window_size(&mut self) -> usize {
657 match self {
658 Self::Ring(s) => s.buffer.drop_to_window_size(),
659 Self::Flat(s) => s.buffer.drop_to_window_size(),
660 }
661 }
662
663 /// Drop exactly `n` bytes from the front of the visible output without
664 /// producing them. Used by `decode_blocks_partial` to discard the
665 /// leading blocks' window-context bytes once the in-range blocks are
666 /// decoded (match resolution complete), leaving only the in-range output.
667 #[cfg(feature = "lsm")]
668 fn buffer_discard_front(&mut self, n: usize) {
669 match self {
670 Self::Ring(s) => s.buffer.discard_front(n),
671 Self::Flat(s) => s.buffer.discard_front(n),
672 }
673 }
674
675 /// Prime the match window with the caller's already-decompressed tail for
676 /// a resumed partial decode. Routes through whichever backend the current
677 /// scratch holds. See [`DecodeBuffer::prime_window`].
678 #[cfg(feature = "lsm")]
679 fn prime_window(&mut self, prefix: &[u8], total_output: u64) {
680 match self {
681 Self::Ring(s) => s.buffer.prime_window(prefix, total_output),
682 Self::Flat(s) => s.buffer.prime_window(prefix, total_output),
683 }
684 }
685
686 /// Total decompressed bytes produced so far (the buffer's running output
687 /// counter, unaffected by window drops / drains). Used to stamp a captured
688 /// [`ResumeState`]'s `output_offset`.
689 #[cfg(feature = "lsm")]
690 fn total_output(&self) -> u64 {
691 match self {
692 Self::Ring(s) => s.buffer.total_output(),
693 Self::Flat(s) => s.buffer.total_output(),
694 }
695 }
696
697 /// Clone the cross-block entropy/repcode state (FSE + Huffman tables +
698 /// `offset_hist`) out of the live scratch for a [`ResumeState`] snapshot.
699 #[cfg(feature = "lsm")]
700 fn export_entropy(
701 &self,
702 dict: Option<&crate::decoding::dictionary::Dictionary>,
703 ) -> (
704 crate::decoding::scratch::FSEScratch,
705 crate::decoding::scratch::HuffmanScratch,
706 [u32; 3],
707 ) {
708 let (fse_src, huf_src, offset_hist) = match self {
709 Self::Ring(s) => (&s.fse, &s.huf, s.offset_hist),
710 Self::Flat(s) => (&s.fse, &s.huf, s.offset_hist),
711 };
712 // The live scratch may still be `Dict`-sourced; `reinit_from` /
713 // `reinit_resolved_from` resolve those axes through the borrow into a
714 // self-contained `Local` snapshot, so the dictionary is required here
715 // whenever the captured frame is dict-backed.
716 let mut fse = crate::decoding::scratch::FSEScratch::new();
717 fse.reinit_from(fse_src, dict);
718 let mut huf = crate::decoding::scratch::HuffmanScratch::new();
719 huf.reinit_resolved_from(huf_src, dict);
720 (fse, huf, offset_hist)
721 }
722
723 /// Install entropy/repcode state from a [`ResumeState`] into the live
724 /// scratch so a `Repeat_Mode` / treeless resume block resolves against the
725 /// same tables a contiguous decode would have carried over.
726 #[cfg(feature = "lsm")]
727 fn restore_entropy(&mut self, state: &ResumeState) {
728 // The `ResumeState` snapshot is a self-contained `Local` materialization
729 // (export resolved every Dict axis into local bytes and detached), so no
730 // dictionary borrow is needed to install it back.
731 match self {
732 Self::Ring(s) => {
733 s.fse.reinit_from(&state.fse, None);
734 s.huf.reinit_resolved_from(&state.huf, None);
735 s.offset_hist = state.offset_hist;
736 }
737 Self::Flat(s) => {
738 s.fse.reinit_from(&state.fse, None);
739 s.huf.reinit_resolved_from(&state.huf, None);
740 s.offset_hist = state.offset_hist;
741 }
742 }
743 }
744
745 /// XXH64 of the window-prime bytes for a [`ResumeState`]: the last
746 /// `min(window_size, buffer_len)` bytes of the current buffer, which at emit
747 /// time are exactly the match-window context the resume block will see.
748 /// Wrap-aware via `last_n_as_slices` — streaming XXH64 over the two slices
749 /// equals a single hash over the contiguous `window_prime` at resume.
750 #[cfg(all(feature = "lsm", feature = "hash"))]
751 fn window_tail_hash(&self, window_size: usize) -> u64 {
752 use core::hash::Hasher;
753 let n = core::cmp::min(window_size, self.buffer_len());
754 let (s1, s2) = self.last_n_as_slices(n);
755 let mut h = twox_hash::XxHash64::with_seed(0);
756 h.write(s1);
757 h.write(s2);
758 h.finish()
759 }
760
761 fn decode_block_content<R: Read>(
762 &mut self,
763 decoder: &mut BlockDecoder,
764 header: &crate::blocks::block::BlockHeader,
765 source: R,
766 dict: Option<&crate::decoding::dictionary::Dictionary>,
767 ) -> Result<u64, DecodeBlockContentError> {
768 match self {
769 Self::Ring(s) => decoder.decode_block_content(header, s, dict, source),
770 Self::Flat(s) => decoder.decode_block_content(header, s, dict, source),
771 }
772 }
773
774 #[cfg(feature = "hash")]
775 fn hash_finish(&self) -> u64 {
776 use core::hash::Hasher;
777 match self {
778 Self::Ring(s) => s.buffer.hash.finish(),
779 Self::Flat(s) => s.buffer.hash.finish(),
780 }
781 }
782
783 /// Forward the drain-time hash toggle to the inner `DecodeBuffer`
784 /// (streaming path). Called by the frame layer from the decoder's
785 /// `ContentChecksum` mode before each decode.
786 #[cfg(feature = "hash")]
787 fn set_compute_hash(&mut self, compute: bool) {
788 match self {
789 Self::Ring(s) => s.buffer.set_compute_hash(compute),
790 Self::Flat(s) => s.buffer.set_compute_hash(compute),
791 }
792 }
793}
794
795struct FrameDecoderState {
796 pub frame_header: frame::FrameHeader,
797 decoder_scratch: DecoderScratchKind,
798 frame_finished: bool,
799 block_counter: usize,
800 bytes_read_counter: u64,
801 check_sum: Option<u32>,
802 using_dict: Option<u32>,
803 /// The dictionary handle applied to this frame, owned for the frame's whole
804 /// decode so the block loop can hand the scratch a `&Dictionary` borrow at
805 /// every `Dict`-sourced table read. ONE refcount clone per dict-apply (not
806 /// per block, not per frame on the reuse path — the same handle stays held
807 /// across `init_with_dict_handle` -> `decode_blocks`); the decode loop
808 /// borrows from this field (disjoint from `decoder_scratch`) with zero
809 /// further clones. `None` on the no-dict path. Cleared by `reset`.
810 active_dict: Option<DictionaryHandle>,
811}
812
813pub enum BlockDecodingStrategy {
814 All,
815 UptoBlocks(usize),
816 UptoBytes(usize),
817}
818
819/// Outcome of [`FrameDecoder::decode_blocks_partial`]: the decompressed
820/// bytes of the requested inner-block range plus where (if anywhere)
821/// decoding stopped early.
822///
823/// Behind the `lsm` Cargo feature.
824#[cfg(feature = "lsm")]
825#[derive(Debug)]
826pub struct PartialDecode {
827 /// Decompressed bytes of the in-range blocks actually decoded, in
828 /// frame order, as one contiguous buffer. `data.len()` equals the sum
829 /// of the decompressed sizes of blocks `start_block .. start_block +
830 /// blocks_decoded`.
831 pub data: alloc::vec::Vec<u8>,
832 /// First block whose output is in [`data`](Self::data): the requested
833 /// `start_block` on a fresh decode, or [`ResumeState::block_index`] when
834 /// resuming (the caller-supplied `start_block` is ignored in resume mode).
835 pub start_block: u32,
836 /// Number of in-range blocks successfully decoded into
837 /// [`data`](Self::data).
838 pub blocks_decoded: u32,
839 /// `Some((block_index, error))` if decoding stopped on a failing block
840 /// before reaching `end_block` (a corrupt block inside the range, or a
841 /// leading block needed for window context). `None` if the requested
842 /// range decoded cleanly or the frame's last block was reached first.
843 ///
844 /// When the failing block is a leading context block
845 /// (`block_index < start_block`), the in-range window could not be
846 /// built so [`data`](Self::data) is empty and `blocks_decoded` is 0.
847 pub stopped_at: Option<(u32, FrameDecoderError)>,
848 /// `true` if the frame's last block was reached during this decode.
849 pub frame_finished: bool,
850 /// Cross-block carry-over state for resuming the next extent. Feed it back
851 /// (with the matching `window_prime`) via the `resume` argument of a
852 /// later [`FrameDecoder::decode_blocks_partial`] to continue from
853 /// [`ResumeState::block_index`] without re-decompressing the prefix.
854 ///
855 /// `None` in two cases: emission was not requested (`emit_resume = false`),
856 /// OR this decode reached the frame's last block ([`frame_finished`] is
857 /// `true`) — there is no following block to resume from, so no snapshot is
858 /// emitted even with `emit_resume = true`. Callers walking a frame
859 /// incrementally should therefore stop when `frame_finished` is set rather
860 /// than treat a `None` here as "emission disabled".
861 ///
862 /// [`frame_finished`]: Self::frame_finished
863 pub resume_state: Option<ResumeState>,
864}
865
866impl FrameDecoderState {
867 /// Window size to actually reserve for this frame's decode buffer.
868 /// A declared content size caps the useful window: matches can never
869 /// reference further back than the bytes that will ever exist, so an
870 /// encoder-declared window above the FCS (e.g. a level-preset window
871 /// on a smaller input) must not inflate the reservation. Every
872 /// `reserve_buffer` site routes through this so the cap is uniform
873 /// across `decode_all_impl`, `decode_blocks`, and the partial path.
874 fn useful_window_size(&self) -> usize {
875 let window_size = self.frame_header.window_size().unwrap_or(0);
876 if self.frame_header.fcs_declared() {
877 window_size.min(self.frame_header.frame_content_size()) as usize
878 } else {
879 window_size as usize
880 }
881 }
882
883 /// The most a frame decoding through the buffer holds, which its growth
884 /// stops at. A single-segment frame's buffer holds its whole content,
885 /// which is its window. A multi-segment frame's ring holds the
886 /// content-capped window plus room for the next block, since each
887 /// compressed block reserves a whole block before it decodes: without it
888 /// the first block past a full window grew the ring and copied the window
889 /// across, and a limit at the content left no block of room once the
890 /// window filled. Upstream sizes its stream buffer as window + block too
891 /// (`ZSTD_decodingBufferSize_min`); it can cap at the content because its
892 /// buffer is not a ring.
893 ///
894 /// A declared content size does NOT cap this. A frame can declare less than
895 /// its blocks go on to produce, and that is caught by the check against the
896 /// declaration once the bytes exist; a limit that stopped the ring short of
897 /// them would instead have the write run out of buffer, which the ring
898 /// asserts on rather than reports. Only the up-front reservation takes the
899 /// declaration ([`Self::decoding_buffer_size`]).
900 fn decoding_buffer_limit(&self) -> usize {
901 let useful_window = self.useful_window_size();
902 if self.frame_header.descriptor.single_segment_flag() {
903 return useful_window;
904 }
905 let window_size = self.frame_header.window_size().unwrap_or(0) as usize;
906 // No overflow: the window was checked against
907 // `MAXIMUM_ALLOWED_WINDOW_SIZE` when the header was taken.
908 useful_window + window_size.min(crate::common::MAX_BLOCK_SIZE as usize)
909 }
910
911 /// What to reserve up front: the limit, except for a multi-segment frame
912 /// whose declared content fits its window, which reserves just its
913 /// content. Only compressed blocks need the block of room past it, and
914 /// they reserve it themselves; the limit then caps that growth at one
915 /// block. Such frames are rare (encoders mark a frame that fits its window
916 /// single-segment), and a small Raw or RLE one should not pay a block.
917 ///
918 /// A frame that declares its size never reserves past the declaration: the
919 /// block of room is there for what a block still has to produce, and an
920 /// honest frame produces exactly what it promised. A 1 MiB window declaring
921 /// one byte more reserved a whole block of room for that byte. A frame that
922 /// goes on to exceed its declaration grows into the limit and is then caught
923 /// by the check against it.
924 fn decoding_buffer_size(&self) -> usize {
925 let window_size = self.frame_header.window_size().unwrap_or(0);
926 if !self.frame_header.fcs_declared() {
927 return self.decoding_buffer_limit();
928 }
929 let declared = self.frame_header.frame_content_size();
930 if declared <= window_size {
931 return self.useful_window_size();
932 }
933 self.decoding_buffer_limit()
934 .min(usize::try_from(declared).unwrap_or(usize::MAX))
935 }
936
937 /// Reserve this frame's decode buffer ([`Self::decoding_buffer_size`])
938 /// and cap its later growth ([`Self::decoding_buffer_limit`]).
939 fn reserve_decoding_buffer(&mut self) {
940 let target = self.decoding_buffer_size();
941 let growth_limit = self.decoding_buffer_limit();
942 // What the frame promised to produce, so the per-block reservation can
943 // ask for the smaller of a block and what is left of that promise.
944 let declared = self
945 .frame_header
946 .fcs_declared()
947 .then(|| self.frame_header.frame_content_size());
948 self.decoder_scratch.set_declared_content(declared);
949 self.decoder_scratch.reserve_buffer(target, growth_limit);
950 }
951
952 /// Construct a new frame decoder state, reading the frame header
953 /// from `source`. When `magicless` is `true`, the 4-byte magic
954 /// number prefix is NOT consumed (upstream zstd `ZSTD_f_zstd1_magicless`).
955 /// Crate-internal — reached only via `FrameDecoder::init` /
956 /// `FrameDecoder::init_with_dict_handle`. The decode buffer is
957 /// allocated lazily on BOTH backends (`new_ring` and `new_flat`):
958 /// direct-eligible frames pay zero buffer allocation, and the
959 /// non-direct fallback reserves `window_size` once in
960 /// `decode_all_impl` / `decode_blocks` via `reserve_buffer` before
961 /// any block write.
962 #[inline]
963 pub(crate) fn new_with_format(
964 source: impl Read,
965 magicless: bool,
966 ) -> Result<FrameDecoderState, FrameDecoderError> {
967 let (frame, header_size) = frame::read_frame_header_with_format(source, magicless)?;
968 Self::new_with_parsed_header(frame, header_size)
969 }
970
971 /// Build a fresh state from an already-parsed frame header (the non-parsing
972 /// tail of [`new_with_format`]). Shared by the `Read` path and the
973 /// slice-direct path ([`FrameDecoder::reset_from_slice`]).
974 pub(crate) fn new_with_parsed_header(
975 frame: frame::FrameHeader,
976 header_size: u8,
977 ) -> Result<FrameDecoderState, FrameDecoderError> {
978 let window_size = frame.window_size()?;
979
980 if window_size > MAXIMUM_ALLOWED_WINDOW_SIZE {
981 return Err(FrameDecoderError::WindowSizeTooBig {
982 requested: window_size,
983 });
984 }
985
986 let decoder_scratch = if frame.descriptor.single_segment_flag() {
987 DecoderScratchKind::new_flat(window_size as usize)
988 } else {
989 DecoderScratchKind::new_ring(window_size as usize)
990 };
991 Ok(FrameDecoderState {
992 frame_header: frame,
993 frame_finished: false,
994 block_counter: 0,
995 decoder_scratch,
996 bytes_read_counter: u64::from(header_size),
997 check_sum: None,
998 using_dict: None,
999 active_dict: None,
1000 })
1001 }
1002
1003 /// Reset this state for a new frame read from `source`, reusing
1004 /// existing allocations. When `magicless` is `true`, the frame
1005 /// header is read WITHOUT expecting a magic-number prefix
1006 /// (upstream zstd `ZSTD_f_zstd1_magicless`). Crate-internal — reached
1007 /// only via `FrameDecoder::reset`.
1008 ///
1009 /// `DecodeBuffer::reset` no longer reserves window_size for either
1010 /// backend — capacity decisions live one layer up. Both backends are
1011 /// lazy: direct-eligible frames pay zero backing-buffer allocation
1012 /// here (they write through `UserSliceBackend`), and the non-direct
1013 /// path is pre-reserved by `decode_all_impl` / `decode_blocks` via
1014 /// `DecoderScratchKind::reserve_buffer(window_size)` before any block
1015 /// write. A reused scratch whose new frame fits within prior capacity
1016 /// reuses it; a larger one grows on that same `reserve_buffer` call.
1017 #[inline]
1018 pub(crate) fn reset_with_format(
1019 &mut self,
1020 source: impl Read,
1021 magicless: bool,
1022 ) -> Result<(), FrameDecoderError> {
1023 let (frame_header, header_size) = frame::read_frame_header_with_format(source, magicless)?;
1024 self.reset_with_parsed_header(frame_header, header_size)
1025 }
1026
1027 /// Apply an already-parsed frame header to this state (the non-parsing tail
1028 /// of [`reset_with_format`]). Shared by the `Read` path and the slice-direct
1029 /// path ([`FrameDecoder::reset_from_slice`]).
1030 #[inline]
1031 pub(crate) fn reset_with_parsed_header(
1032 &mut self,
1033 frame_header: frame::FrameHeader,
1034 header_size: u8,
1035 ) -> Result<(), FrameDecoderError> {
1036 let window_size = frame_header.window_size()?;
1037
1038 if window_size > MAXIMUM_ALLOWED_WINDOW_SIZE {
1039 return Err(FrameDecoderError::WindowSizeTooBig {
1040 requested: window_size,
1041 });
1042 }
1043
1044 self.decoder_scratch
1045 .reset(&frame_header, window_size as usize);
1046 self.frame_header = frame_header;
1047 self.frame_finished = false;
1048 self.block_counter = 0;
1049 self.bytes_read_counter = u64::from(header_size);
1050 self.check_sum = None;
1051 self.using_dict = None;
1052 // `active_dict` is intentionally NOT cleared here: it is only ever READ
1053 // while a scratch table source is `Dict`, which `init_from_dict` arms
1054 // on a dict frame and which a no-dict frame leaves `Local` (so a stale
1055 // held handle is never read). Keeping it lets the per-apply `ptr::eq`
1056 // reuse-check below skip the clone when the SAME dictionary is
1057 // re-applied frame-over-frame (the CoordiNode per-label-dict hot path)
1058 // — zero refcount churn on reuse, one clone only on a genuine swap.
1059 Ok(())
1060 }
1061
1062 /// Hold the dictionary handle for this frame's whole decode so the block
1063 /// loop can borrow `&Dictionary` at every `Dict`-sourced read. Clones the
1064 /// handle ONLY when it is a different dictionary than the one already held
1065 /// (`ptr::eq` on the `Arc`'s pointee) — so re-applying the SAME dictionary
1066 /// frame-over-frame (the reuse hot path) costs zero refcount churn.
1067 fn set_active_dict(&mut self, dict: &DictionaryHandle) {
1068 if self
1069 .active_dict
1070 .as_ref()
1071 .is_none_or(|held| !core::ptr::eq(held.as_dict(), dict.as_dict()))
1072 {
1073 self.active_dict = Some(dict.clone());
1074 }
1075 }
1076}
1077
1078impl Default for FrameDecoder {
1079 fn default() -> Self {
1080 Self::new()
1081 }
1082}
1083
1084impl FrameDecoder {
1085 /// This will create a new decoder without allocating anything yet.
1086 /// init()/reset() will allocate all needed buffers if it is the first time this decoder is used
1087 /// else they just reset these buffers with not further allocations
1088 pub fn new() -> FrameDecoder {
1089 FrameDecoder {
1090 state: None,
1091 #[cfg(test)]
1092 direct_frames: 0,
1093 owned_dicts: BTreeMap::new(),
1094 #[cfg(target_has_atomic = "ptr")]
1095 shared_dicts: BTreeMap::new(),
1096 #[cfg(not(target_has_atomic = "ptr"))]
1097 shared_dicts: (),
1098 magicless: false,
1099 content_checksum: ContentChecksum::EmitOnly,
1100 #[cfg(feature = "lsm")]
1101 expect_dict_id: None,
1102 #[cfg(feature = "lsm")]
1103 expect_window_descriptor: None,
1104 #[cfg(all(feature = "lsm", feature = "hash"))]
1105 per_block_checksums_enabled: false,
1106 #[cfg(all(feature = "lsm", feature = "hash"))]
1107 computed_block_checksums: alloc::vec::Vec::new(),
1108 kernel: crate::cpu_kernel::detect_cpu_kernel(),
1109 }
1110 }
1111
1112 /// Decode with a named CPU tier instead of the one this machine selected.
1113 ///
1114 /// Which tier runs is otherwise a property of the host, so a test on any
1115 /// one machine covers exactly one of them. The tiers are separate
1116 /// monomorphs with their own gates and their own argument plumbing, and
1117 /// a divergence between them is precisely what no single-host run can see.
1118 ///
1119 /// The caller is responsible for naming a tier the CPU can actually
1120 /// execute: the monoliths carry `target_feature`, and asking for one the
1121 /// machine lacks is undefined behaviour rather than an error.
1122 #[cfg(test)]
1123 // Its consumer is the per-tier agreement test, and the tiers worth
1124 // disagreeing are the x86 ones, so on other architectures nothing calls it.
1125 #[allow(dead_code)]
1126 pub(crate) fn decode_with_kernel_for_tests(&mut self, kernel: crate::cpu_kernel::CpuKernelTag) {
1127 self.kernel = kernel;
1128 }
1129
1130 /// Heap bytes currently held by the decoder's lazily-grown workspace:
1131 /// the decode-window buffer plus the per-block literal/content buffers
1132 /// and the entropy tables. Returns 0 before the first frame is initialised
1133 /// (no workspace allocated yet). The window allocation dominates and grows
1134 /// with the frame's window size; this is the value to track for decode-time
1135 /// memory pressure, mirroring the workspace term of upstream
1136 /// `ZSTD_sizeof_DCtx`. Shared dictionaries (ref-counted handles) are not
1137 /// counted, matching upstream excluding `refDDict` memory.
1138 pub fn workspace_size(&self) -> usize {
1139 self.state
1140 .as_ref()
1141 .map_or(0, |s| s.decoder_scratch.workspace_bytes())
1142 }
1143
1144 /// Select how the frame's optional content checksum is handled
1145 /// (compute, expose, verify, or skip). See [`ContentChecksum`].
1146 /// Default [`ContentChecksum::EmitOnly`]. Takes effect on the next
1147 /// decode; safe to call between frames on a reused decoder.
1148 pub fn set_content_checksum(&mut self, mode: ContentChecksum) {
1149 self.content_checksum = mode;
1150 }
1151
1152 /// Opt in to per-block XXH64 verification during decode.
1153 /// Default off; zero cost when disabled. Each block's decompressed
1154 /// bytes are XXH64-hashed (low 32 bits) and appended to
1155 /// [`Self::computed_block_checksums`] as the decode progresses.
1156 /// Callers compare the captured digests against externally-stored
1157 /// expected values (e.g. from a per-block sidecar in the
1158 /// containing application protocol).
1159 ///
1160 /// Behind `all(feature = "lsm", feature = "hash")` — the XXH64
1161 /// primitive lives behind the `hash` feature, so this method
1162 /// only compiles when both are enabled.
1163 #[cfg(all(feature = "lsm", feature = "hash"))]
1164 pub fn enable_per_block_checksums(&mut self) {
1165 self.per_block_checksums_enabled = true;
1166 }
1167
1168 /// Per-block XXH64 (low 32 bits) digests captured during the
1169 /// current frame's decode. Empty unless
1170 /// [`Self::enable_per_block_checksums`] was called before
1171 /// [`Self::decode_all`] / [`Self::reset`].
1172 ///
1173 /// Reset at the start of every new frame.
1174 ///
1175 /// Behind `all(feature = "lsm", feature = "hash")`.
1176 #[cfg(all(feature = "lsm", feature = "hash"))]
1177 pub fn computed_block_checksums(&self) -> &[u32] {
1178 &self.computed_block_checksums
1179 }
1180
1181 /// Pin the expected `Dictionary_ID` for the next frame.
1182 ///
1183 /// When `expected` is set, [`Self::init`] / [`Self::reset`]
1184 /// validate it against the parsed frame header BEFORE any
1185 /// block decode work runs. A mismatch returns
1186 /// [`crate::decoding::errors::FrameDecoderError::UnexpectedDictId`]
1187 /// before any block decode and before any output is produced.
1188 /// Scratch buffer allocation / reservation for the decode
1189 /// pipeline happens during frame-header parsing, which is
1190 /// already complete when this validation fires — the cost of
1191 /// scratch sizing is paid even on a mismatched header. The
1192 /// guarantee is "no block decode, no XXH64 init, no partial
1193 /// output", not "zero allocation".
1194 ///
1195 /// `Some(0)` is treated as "no dictionary expected": a frame
1196 /// whose header omits the optional `Dictionary_ID` field
1197 /// (flag value 0) passes the check; a frame that carries an
1198 /// explicit non-zero id fails.
1199 ///
1200 /// `None` (default) disables the check.
1201 ///
1202 /// Primary use case: post-AEAD-decrypt sanity check in
1203 /// wire-format consumers (e.g. lsm-tree's encrypted block
1204 /// format pins the `dict_id` baked into the AAD against the
1205 /// inner zstd frame's `dict_id` to defeat dict-substitution
1206 /// attacks).
1207 ///
1208 /// NOT a replacement for AEAD authentication. NOT the same
1209 /// semantic as upstream zstd `ZSTD_d_windowLogMax` (which is a
1210 /// ceiling-style limit, separate concern).
1211 #[cfg(feature = "lsm")]
1212 #[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
1213 pub fn expect_dict_id(&mut self, expected: Option<u32>) {
1214 self.expect_dict_id = expected;
1215 }
1216
1217 /// Pin the expected raw `Window_Descriptor` byte (RFC 8878
1218 /// §3.1.1.1.2 layout: `(exp << 3) | mantissa`) for the next
1219 /// frame.
1220 ///
1221 /// When `expected` is set, [`Self::init`] / [`Self::reset`]
1222 /// validate it against the parsed frame header BEFORE any
1223 /// block decode work runs. A mismatch returns
1224 /// [`crate::decoding::errors::FrameDecoderError::UnexpectedWindowDescriptor`].
1225 ///
1226 /// Single-segment frames omit the `Window_Descriptor` byte
1227 /// from the wire entirely. Setting an expectation while
1228 /// receiving a single-segment frame fails the check with
1229 /// `found: None` — there is no on-wire byte to match against,
1230 /// which is reported explicitly rather than silently passing.
1231 ///
1232 /// `None` (default) disables the check.
1233 ///
1234 /// Byte-exact equality, NOT a ceiling. Upstream zstd
1235 /// `ZSTD_d_windowLogMax` is a separate ceiling-style limit
1236 /// available through the C FFI surface; this method is for
1237 /// strict equality validation against a pinned expectation
1238 /// (e.g. lsm-tree's wire format pins the window descriptor
1239 /// from the AAD to defeat decompression-bomb-swap attacks).
1240 #[cfg(feature = "lsm")]
1241 #[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
1242 pub fn expect_window_descriptor(&mut self, expected: Option<u8>) {
1243 self.expect_window_descriptor = expected;
1244 }
1245
1246 /// Validate the just-parsed frame header against any pinned
1247 /// expectations set via [`Self::expect_dict_id`] /
1248 /// [`Self::expect_window_descriptor`].
1249 ///
1250 /// Returns the typed error variant on mismatch and leaves
1251 /// `self.state` in a re-resettable shape — a subsequent
1252 /// `reset()` will overwrite `frame_header` from the new source
1253 /// without needing intermediate cleanup.
1254 #[cfg(feature = "lsm")]
1255 fn validate_expectations(
1256 &self,
1257 frame_header: &frame::FrameHeader,
1258 ) -> Result<(), FrameDecoderError> {
1259 if let Some(expected) = self.expect_dict_id {
1260 let found = frame_header.dictionary_id();
1261 // `Some(0)` is the "no dictionary expected" sentinel —
1262 // matches a frame whose header omits the optional
1263 // dict_id field (which is reported as `None` by the
1264 // parser). All other values must match exactly.
1265 let matches = match (expected, found) {
1266 (0, None) => true,
1267 (e, Some(f)) => e == f,
1268 _ => false,
1269 };
1270 if !matches {
1271 return Err(FrameDecoderError::UnexpectedDictId {
1272 expected: Some(expected),
1273 found,
1274 });
1275 }
1276 }
1277 if let Some(expected) = self.expect_window_descriptor {
1278 let found = frame_header.window_descriptor();
1279 if found != Some(expected) {
1280 return Err(FrameDecoderError::UnexpectedWindowDescriptor { expected, found });
1281 }
1282 }
1283 Ok(())
1284 }
1285
1286 /// Enable or disable magicless frame format
1287 /// (`ZSTD_f_zstd1_magicless`). When set to `true`, subsequent
1288 /// [`init`] / [`reset`] calls expect the frame header to begin
1289 /// directly with the frame-header descriptor — no 4-byte magic
1290 /// number prefix. Default false. Must match the encoder's
1291 /// magicless setting; the format is unambiguous only when the
1292 /// caller knows it out-of-band.
1293 ///
1294 /// Note: magicless mode also disables skippable-frame detection.
1295 /// The `0x184D2A50..=0x184D2A5F` skippable-frame magic range is
1296 /// only recognised when the 4-byte magic prefix is consumed, so
1297 /// `decode_all` / `init` / `reset` will treat a skippable frame
1298 /// at the head of a magicless stream as a malformed frame header
1299 /// (bad descriptor / window-size error) instead of skipping it.
1300 /// Mixed-format streams that interleave skippable frames must be
1301 /// pre-split by the caller; `set_magicless(true)` is only safe
1302 /// when the entire stream is known to be magicless zstd frames.
1303 pub fn set_magicless(&mut self, magicless: bool) {
1304 self.magicless = magicless;
1305 }
1306
1307 #[cfg(target_has_atomic = "ptr")]
1308 fn shared_dict_exists(&self, dict_id: u32) -> bool {
1309 self.shared_dicts.contains_key(&dict_id)
1310 }
1311
1312 #[cfg(not(target_has_atomic = "ptr"))]
1313 fn shared_dict_exists(&self, _dict_id: u32) -> bool {
1314 false
1315 }
1316
1317 fn validate_registered_dictionary(dict: &Dictionary) -> Result<(), FrameDecoderError> {
1318 use crate::decoding::errors::DictionaryDecodeError as dict_err;
1319
1320 // Registration keys the dictionary by its ID, so a zero one has no
1321 // slot to occupy and no frame could ever select it.
1322 if dict.id == 0 {
1323 return Err(FrameDecoderError::from(dict_err::ZeroDictionaryId));
1324 }
1325 Self::validate_dictionary_content(dict)
1326 }
1327
1328 /// Checks a dictionary can be used at all, whatever route it arrived by.
1329 ///
1330 /// Separate from [`Self::validate_registered_dictionary`] because a
1331 /// dictionary supplied explicitly needs no ID: a raw-content dictionary
1332 /// has no header to carry one, and the frames built with it record none.
1333 fn validate_dictionary_content(dict: &Dictionary) -> Result<(), FrameDecoderError> {
1334 use crate::decoding::errors::DictionaryDecodeError as dict_err;
1335
1336 if let Some(index) = dict.offset_hist.iter().position(|&rep| rep == 0) {
1337 return Err(FrameDecoderError::from(
1338 dict_err::ZeroRepeatOffsetInDictionary { index: index as u8 },
1339 ));
1340 }
1341 Ok(())
1342 }
1343
1344 /// init() will allocate all needed buffers if it is the first time this decoder is used
1345 /// else they just reset these buffers with not further allocations
1346 ///
1347 /// Note that all bytes currently in the decodebuffer from any previous frame will be lost. Collect them with collect()/collect_to_writer()
1348 ///
1349 /// equivalent to reset()
1350 #[inline]
1351 pub fn init(&mut self, source: impl Read) -> Result<(), FrameDecoderError> {
1352 self.reset(source)
1353 }
1354
1355 /// Initialize the decoder for a new frame using a pre-parsed dictionary handle.
1356 ///
1357 /// If the frame header has a dictionary ID, this validates it against
1358 /// `dict.id()` and returns [`FrameDecoderError::DictIdMismatch`] on mismatch.
1359 ///
1360 /// If the header omits the optional dictionary ID, this still applies the
1361 /// provided dictionary handle.
1362 ///
1363 /// # Warning
1364 ///
1365 /// This method always applies `dict` unless the frame header contains a
1366 /// non-matching dictionary ID. Callers must only use this API when they
1367 /// already know the frame was encoded with the provided dictionary, even if
1368 /// the frame header omits the dictionary ID or encodes an explicit
1369 /// dictionary ID of `0`.
1370 ///
1371 /// Passing a dictionary for a frame that was not encoded with it can
1372 /// silently corrupt the decoded output.
1373 pub fn init_with_dict_handle(
1374 &mut self,
1375 source: impl Read,
1376 dict: &DictionaryHandle,
1377 ) -> Result<(), FrameDecoderError> {
1378 self.reset_with_dict_handle(source, dict)
1379 }
1380
1381 /// reset() will allocate all needed buffers if it is the first time this decoder is used
1382 /// else they just reset these buffers with not further allocations
1383 ///
1384 /// Note that all bytes currently in the decodebuffer from any previous frame will be lost. Collect them with collect()/collect_to_writer()
1385 ///
1386 /// equivalent to init()
1387 #[inline]
1388 pub fn reset(&mut self, source: impl Read) -> Result<(), FrameDecoderError> {
1389 use FrameDecoderError as err;
1390 // Fresh frame → start with an empty per-block checksum vec so
1391 // the values for the next frame don't carry over from the
1392 // previous one.
1393 #[cfg(all(feature = "lsm", feature = "hash"))]
1394 self.computed_block_checksums.clear();
1395 let magicless = self.magicless;
1396 let dict_id = match &mut self.state {
1397 Some(s) => {
1398 s.reset_with_format(source, magicless)?;
1399 s.frame_header.dictionary_id()
1400 }
1401 None => {
1402 self.state = Some(FrameDecoderState::new_with_format(source, magicless)?);
1403 self.state
1404 .as_ref()
1405 .and_then(|state| state.frame_header.dictionary_id())
1406 }
1407 };
1408 // Validate any pinned expectations BEFORE block decode work
1409 // runs. Catches dict_id substitution / window-descriptor
1410 // tampering on inputs already authenticated by an outer
1411 // layer (e.g. AEAD). Returning here leaves `self.state` in
1412 // a re-resettable shape — next `reset()` re-parses the
1413 // frame header without intermediate cleanup.
1414 #[cfg(feature = "lsm")]
1415 if let Some(state) = self.state.as_ref() {
1416 self.validate_expectations(&state.frame_header)?;
1417 }
1418 if let Some(dict_id) = dict_id {
1419 let state = self.state.as_mut().expect("state initialized");
1420 let owned_dicts = &self.owned_dicts;
1421 #[cfg(target_has_atomic = "ptr")]
1422 let shared_dicts = &self.shared_dicts;
1423 let dict = owned_dicts
1424 .get(&dict_id)
1425 .or_else(|| {
1426 #[cfg(target_has_atomic = "ptr")]
1427 {
1428 shared_dicts.get(&dict_id)
1429 }
1430 #[cfg(not(target_has_atomic = "ptr"))]
1431 {
1432 None
1433 }
1434 })
1435 .ok_or(err::DictNotProvided { dict_id })?;
1436 state.decoder_scratch.init_from_dict(dict);
1437 state.set_active_dict(dict);
1438 state.using_dict = Some(dict_id);
1439 }
1440 Ok(())
1441 }
1442
1443 /// Slice-direct equivalent of [`reset`](Self::reset) for the in-memory
1444 /// decode path: parses the frame header straight out of `*input` via
1445 /// [`frame::read_frame_header_from_slice`] (no `Read`-trait `read_exact`
1446 /// per field) and advances `*input` past it, then applies it through the
1447 /// shared parsed-header path. Behaviour — including skippable-frame and
1448 /// truncation errors, dictionary-id resolution, and pinned-expectation
1449 /// validation — is identical to `reset`; only the header read avoids the
1450 /// `io::impls` dispatch.
1451 pub(crate) fn reset_from_slice(&mut self, input: &mut &[u8]) -> Result<(), FrameDecoderError> {
1452 use FrameDecoderError as err;
1453 #[cfg(all(feature = "lsm", feature = "hash"))]
1454 self.computed_block_checksums.clear();
1455 let magicless = self.magicless;
1456 let (frame_header, header_size) = frame::read_frame_header_from_slice(input, magicless)?;
1457 let dict_id = match &mut self.state {
1458 Some(s) => {
1459 s.reset_with_parsed_header(frame_header, header_size)?;
1460 s.frame_header.dictionary_id()
1461 }
1462 None => {
1463 self.state = Some(FrameDecoderState::new_with_parsed_header(
1464 frame_header,
1465 header_size,
1466 )?);
1467 self.state
1468 .as_ref()
1469 .and_then(|state| state.frame_header.dictionary_id())
1470 }
1471 };
1472 #[cfg(feature = "lsm")]
1473 if let Some(state) = self.state.as_ref() {
1474 self.validate_expectations(&state.frame_header)?;
1475 }
1476 if let Some(dict_id) = dict_id {
1477 let state = self.state.as_mut().expect("state initialized");
1478 let owned_dicts = &self.owned_dicts;
1479 #[cfg(target_has_atomic = "ptr")]
1480 let shared_dicts = &self.shared_dicts;
1481 let dict = owned_dicts
1482 .get(&dict_id)
1483 .or_else(|| {
1484 #[cfg(target_has_atomic = "ptr")]
1485 {
1486 shared_dicts.get(&dict_id)
1487 }
1488 #[cfg(not(target_has_atomic = "ptr"))]
1489 {
1490 None
1491 }
1492 })
1493 .ok_or(err::DictNotProvided { dict_id })?;
1494 state.decoder_scratch.init_from_dict(dict);
1495 state.set_active_dict(dict);
1496 state.using_dict = Some(dict_id);
1497 }
1498 Ok(())
1499 }
1500
1501 /// Reset this decoder for a new frame using a pre-parsed dictionary handle.
1502 ///
1503 /// If the frame header has a dictionary ID, this validates it against
1504 /// `dict.id()` and returns [`FrameDecoderError::DictIdMismatch`] on mismatch.
1505 ///
1506 /// If the header omits the optional dictionary ID, this still applies the
1507 /// provided dictionary handle.
1508 ///
1509 /// # Warning
1510 ///
1511 /// This method always applies `dict` unless the frame header contains a
1512 /// non-matching dictionary ID. Callers must only use this API when they
1513 /// already know the frame was encoded with the provided dictionary, even if
1514 /// the frame header omits the dictionary ID or encodes an explicit
1515 /// dictionary ID of `0`.
1516 ///
1517 /// Passing a dictionary for a frame that was not encoded with it can
1518 /// silently corrupt the decoded output.
1519 pub fn reset_with_dict_handle(
1520 &mut self,
1521 source: impl Read,
1522 dict: &DictionaryHandle,
1523 ) -> Result<(), FrameDecoderError> {
1524 self.reset_frame_for_dict(source, dict)?;
1525 self.state
1526 .as_mut()
1527 .expect("state populated by reset_frame_for_dict")
1528 .set_active_dict(dict);
1529 Ok(())
1530 }
1531
1532 /// [`reset_with_dict_handle`](Self::reset_with_dict_handle) with the
1533 /// dictionary the decode state already holds, for the frames that follow
1534 /// a forced-dictionary frame in one stream. The handle is moved out and
1535 /// back rather than passed in, so no reference count is touched.
1536 pub(crate) fn reset_with_active_dict(
1537 &mut self,
1538 source: impl Read,
1539 ) -> Result<(), FrameDecoderError> {
1540 let dict = self
1541 .state
1542 .as_mut()
1543 .and_then(|state| state.active_dict.take())
1544 .expect("a forced-dictionary stream's decoder holds its dictionary");
1545 let reset = self.reset_frame_for_dict(source, &dict);
1546 // Back in place whatever the outcome, so a failed header leaves the
1547 // decoder holding its dictionary as before.
1548 self.state
1549 .as_mut()
1550 .expect("the state the dictionary was taken from")
1551 .active_dict = Some(dict);
1552 reset
1553 }
1554
1555 /// Everything [`reset_with_dict_handle`](Self::reset_with_dict_handle)
1556 /// does except storing the handle: parse the header, check the ID, and
1557 /// point the entropy tables at `dict`.
1558 fn reset_frame_for_dict(
1559 &mut self,
1560 source: impl Read,
1561 dict: &DictionaryHandle,
1562 ) -> Result<(), FrameDecoderError> {
1563 use FrameDecoderError as err;
1564 // Fresh frame → drop the previous frame's per-block checksum
1565 // digests so the next decode starts with an empty vec.
1566 // Mirrors the same clear in `reset()`; reset_with_dict_handle
1567 // is a parallel entry point so it needs its own call.
1568 #[cfg(all(feature = "lsm", feature = "hash"))]
1569 self.computed_block_checksums.clear();
1570 Self::validate_dictionary_content(dict.as_dict())?;
1571 let magicless = self.magicless;
1572 // Scope the &mut borrow of `self.state` to the header parse
1573 // alone, so the subsequent `validate_expectations(&self, ...)`
1574 // call below can take a fresh shared borrow of self without
1575 // tripping the borrow checker.
1576 match &mut self.state {
1577 Some(s) => s.reset_with_format(source, magicless)?,
1578 None => {
1579 self.state = Some(FrameDecoderState::new_with_format(source, magicless)?);
1580 }
1581 }
1582 // Single source of truth: route through the same
1583 // `validate_expectations` used by `reset()`. Routing through
1584 // the helper keeps the two code paths from drifting (e.g.,
1585 // if expect-semantics or error wiring changes later).
1586 #[cfg(feature = "lsm")]
1587 {
1588 let header = &self
1589 .state
1590 .as_ref()
1591 .expect("state populated by reset_with_format/new_with_format")
1592 .frame_header;
1593 self.validate_expectations(header)?;
1594 }
1595 let state = self
1596 .state
1597 .as_mut()
1598 .expect("state populated by reset_with_format/new_with_format");
1599 if let Some(dict_id) = state.frame_header.dictionary_id()
1600 && dict_id != dict.id()
1601 {
1602 return Err(err::DictIdMismatch {
1603 expected: dict_id,
1604 provided: dict.id(),
1605 });
1606 }
1607 state.decoder_scratch.init_from_dict(dict);
1608 state.using_dict = Some(dict.id());
1609 Ok(())
1610 }
1611
1612 /// Slice-direct equivalent of [`reset_with_dict_handle`](Self::reset_with_dict_handle)
1613 /// for the in-memory decode path: parses the frame header straight out of
1614 /// `*input` via [`frame::read_frame_header_from_slice`] (no `Read`-trait
1615 /// `read_exact` per field) and applies it through the shared parsed-header
1616 /// path, then attaches `dict`. Behaviour — dictionary-id mismatch, pinned
1617 /// expectations, scratch init — is identical to `reset_with_dict_handle`;
1618 /// only the header read avoids the `io::impls` dispatch.
1619 pub(crate) fn reset_from_slice_with_dict_handle(
1620 &mut self,
1621 input: &mut &[u8],
1622 dict: &DictionaryHandle,
1623 ) -> Result<(), FrameDecoderError> {
1624 use FrameDecoderError as err;
1625 #[cfg(all(feature = "lsm", feature = "hash"))]
1626 self.computed_block_checksums.clear();
1627 Self::validate_dictionary_content(dict.as_dict())?;
1628 let magicless = self.magicless;
1629 let (frame_header, header_size) = frame::read_frame_header_from_slice(input, magicless)?;
1630 match &mut self.state {
1631 Some(s) => s.reset_with_parsed_header(frame_header, header_size)?,
1632 None => {
1633 self.state = Some(FrameDecoderState::new_with_parsed_header(
1634 frame_header,
1635 header_size,
1636 )?);
1637 }
1638 }
1639 #[cfg(feature = "lsm")]
1640 {
1641 let header = &self
1642 .state
1643 .as_ref()
1644 .expect("state populated by reset_with_parsed_header/new_with_parsed_header")
1645 .frame_header;
1646 self.validate_expectations(header)?;
1647 }
1648 let state = self
1649 .state
1650 .as_mut()
1651 .expect("state populated by reset_with_parsed_header/new_with_parsed_header");
1652 if let Some(dict_id) = state.frame_header.dictionary_id()
1653 && dict_id != dict.id()
1654 {
1655 return Err(err::DictIdMismatch {
1656 expected: dict_id,
1657 provided: dict.id(),
1658 });
1659 }
1660 state.decoder_scratch.init_from_dict(dict);
1661 state.set_active_dict(dict);
1662 state.using_dict = Some(dict.id());
1663 Ok(())
1664 }
1665
1666 /// Add a dictionary that can be selected dynamically by frame dictionary ID.
1667 ///
1668 /// Returns [`FrameDecoderError::DictAlreadyRegistered`] if the ID is already
1669 /// registered (either as owned or shared).
1670 pub fn add_dict(&mut self, dict: Dictionary) -> Result<(), FrameDecoderError> {
1671 Self::validate_registered_dictionary(&dict)?;
1672 let dict_id = dict.id;
1673 if self.owned_dicts.contains_key(&dict_id) || self.shared_dict_exists(dict_id) {
1674 return Err(FrameDecoderError::DictAlreadyRegistered { dict_id });
1675 }
1676 self.owned_dicts
1677 .insert(dict_id, DictionaryHandle::from_dictionary(dict));
1678 Ok(())
1679 }
1680
1681 /// Parse and add a serialized dictionary blob.
1682 pub fn add_dict_from_bytes(&mut self, raw_dictionary: &[u8]) -> Result<(), FrameDecoderError> {
1683 let dict = Dictionary::decode_dict(raw_dictionary)?;
1684 self.add_dict(dict)
1685 }
1686
1687 /// Add a pre-parsed dictionary handle for reuse across decoders.
1688 ///
1689 /// This API is available on targets with pointer-width atomics
1690 /// (`target_has_atomic = "ptr"`).
1691 ///
1692 /// Returns [`FrameDecoderError::DictAlreadyRegistered`] if the ID is already
1693 /// registered (either as owned or shared).
1694 #[cfg(target_has_atomic = "ptr")]
1695 pub fn add_dict_handle(&mut self, dict: DictionaryHandle) -> Result<(), FrameDecoderError> {
1696 Self::validate_registered_dictionary(dict.as_dict())?;
1697 let dict_id = dict.id();
1698 if self.owned_dicts.contains_key(&dict_id) || self.shared_dicts.contains_key(&dict_id) {
1699 return Err(FrameDecoderError::DictAlreadyRegistered { dict_id });
1700 }
1701 self.shared_dicts.insert(dict_id, dict);
1702 Ok(())
1703 }
1704
1705 pub fn force_dict(&mut self, dict_id: u32) -> Result<(), FrameDecoderError> {
1706 use FrameDecoderError as err;
1707 let state = self.state.as_mut().ok_or(err::NotYetInitialized)?;
1708 let owned_dicts = &self.owned_dicts;
1709 #[cfg(target_has_atomic = "ptr")]
1710 let shared_dicts = &self.shared_dicts;
1711
1712 let dict = owned_dicts
1713 .get(&dict_id)
1714 .or_else(|| {
1715 #[cfg(target_has_atomic = "ptr")]
1716 {
1717 shared_dicts.get(&dict_id)
1718 }
1719 #[cfg(not(target_has_atomic = "ptr"))]
1720 {
1721 None
1722 }
1723 })
1724 .ok_or(err::DictNotProvided { dict_id })?;
1725 state.decoder_scratch.init_from_dict(dict);
1726 state.set_active_dict(dict);
1727 state.using_dict = Some(dict_id);
1728
1729 Ok(())
1730 }
1731
1732 /// Returns how many bytes the frame contains after decompression
1733 pub fn content_size(&self) -> u64 {
1734 match &self.state {
1735 None => 0,
1736 Some(s) => s.frame_header.frame_content_size(),
1737 }
1738 }
1739
1740 /// Returns the checksum that was read from the data. Only available after all bytes have been read. It is the last 4 bytes of a zstd-frame
1741 pub fn get_checksum_from_data(&self) -> Option<u32> {
1742 let state = self.state.as_ref()?;
1743
1744 state.check_sum
1745 }
1746
1747 /// Returns the checksum that was calculated while decoding.
1748 /// Only a sensible value after all decoded bytes have been collected/read from the FrameDecoder.
1749 /// Returns `None` when the frame header has `content_checksum_flag = 0`:
1750 /// no hash is computed for such frames (the post-decode XXH64 pass was a
1751 /// 63 % decode-wall hotspot on flag-off frames; skipping it when the
1752 /// frame format declares no trailing digest avoids that wasted work).
1753 #[cfg(feature = "hash")]
1754 pub fn get_calculated_checksum(&self) -> Option<u32> {
1755 let state = self.state.as_ref()?;
1756 // `ContentChecksum::None` skips the XXH64 pass entirely, so there is
1757 // no calculated digest to report.
1758 if self.content_checksum == ContentChecksum::None {
1759 return None;
1760 }
1761 if !state.frame_header.descriptor.content_checksum_flag() {
1762 return None;
1763 }
1764 let cksum_64bit = state.decoder_scratch.hash_finish();
1765 //truncate to lower 32bit because reasons...
1766 Some(cksum_64bit as u32)
1767 }
1768
1769 /// Compare the frame's stored content checksum against the digest the
1770 /// decoder computed, returning [`FrameDecoderError::ChecksumMismatch`] on
1771 /// disagreement. No-op unless the mode is [`ContentChecksum::Verify`] and
1772 /// the frame carries a trailing checksum.
1773 ///
1774 /// [`decode_all`](Self::decode_all) and the streaming reader call this
1775 /// automatically. Callers driving [`decode_blocks`](Self::decode_blocks)
1776 /// directly invoke it themselves once per frame, after the frame is fully
1777 /// decoded AND fully drained (e.g. via [`collect`](Self::collect)), so both
1778 /// the stored value and the running digest are final.
1779 #[cfg(feature = "hash")]
1780 pub fn verify_content_checksum(&self) -> Result<(), FrameDecoderError> {
1781 if self.content_checksum != ContentChecksum::Verify {
1782 return Ok(());
1783 }
1784 let Some(state) = self.state.as_ref() else {
1785 return Ok(());
1786 };
1787 if !state.frame_header.descriptor.content_checksum_flag() {
1788 return Ok(());
1789 }
1790 let Some(expected) = state.check_sum else {
1791 return Ok(());
1792 };
1793 let calculated = state.decoder_scratch.hash_finish() as u32;
1794 if expected != calculated {
1795 return Err(FrameDecoderError::ChecksumMismatch {
1796 expected,
1797 calculated,
1798 });
1799 }
1800 Ok(())
1801 }
1802
1803 /// Counter for how many bytes have been consumed while decoding the frame
1804 pub fn bytes_read_from_source(&self) -> u64 {
1805 let state = match &self.state {
1806 None => return 0,
1807 Some(s) => s,
1808 };
1809 state.bytes_read_counter
1810 }
1811
1812 /// Test-only: number of frames decoded through the single-copy direct
1813 /// path (`run_direct_decode`). Lets cross-module tests assert that a
1814 /// given decode took the decode-in-place path rather than the ring drain.
1815 #[cfg(test)]
1816 pub(crate) fn direct_frames(&self) -> u64 {
1817 self.direct_frames
1818 }
1819
1820 /// Test-only: whether the decode state currently holds an owning dictionary
1821 /// handle (`active_dict`). Every path that arms `Dict`-sourced scratch tables
1822 /// must also install this handle, or a later dict-table read resolves `None`.
1823 #[cfg(test)]
1824 pub(crate) fn active_dict_installed(&self) -> bool {
1825 self.state.as_ref().is_some_and(|s| s.active_dict.is_some())
1826 }
1827
1828 /// Whether the current frames last block has been decoded yet
1829 /// If this returns true you can call the drain* functions to get all content
1830 /// (the read() function will drain automatically if this returns true)
1831 pub fn is_finished(&self) -> bool {
1832 let state = match &self.state {
1833 None => return true,
1834 Some(s) => s,
1835 };
1836 if state.frame_header.descriptor.content_checksum_flag() {
1837 state.frame_finished && state.check_sum.is_some()
1838 } else {
1839 state.frame_finished
1840 }
1841 }
1842
1843 /// Counter for how many blocks have already been decoded
1844 pub fn blocks_decoded(&self) -> usize {
1845 let state = match &self.state {
1846 None => return 0,
1847 Some(s) => s,
1848 };
1849 state.block_counter
1850 }
1851
1852 /// Decodes blocks from a reader. It requires that the framedecoder has been initialized first.
1853 /// The Strategy influences how many blocks will be decoded before the function returns
1854 /// This is important if you want to manage memory consumption carefully. If you don't care
1855 /// about that you can just choose the strategy "All" and have all blocks of the frame decoded into the buffer
1856 pub fn decode_blocks(
1857 &mut self,
1858 mut source: impl Read,
1859 strat: BlockDecodingStrategy,
1860 ) -> Result<bool, FrameDecoderError> {
1861 use FrameDecoderError as err;
1862 // Apply the content-checksum mode to the streaming drain hash before
1863 // any block decodes into the ring. Hash only when a digest is both
1864 // wanted (mode != None) AND present in the frame (content_checksum_flag
1865 // set) — a flag-off frame has nothing to verify or expose, so hashing
1866 // it is wasted work. Mirrors the direct path and get_calculated_checksum.
1867 #[cfg(feature = "hash")]
1868 let checksum_mode = self.content_checksum;
1869 let state = self.state.as_mut().ok_or(err::NotYetInitialized)?;
1870 #[cfg(feature = "hash")]
1871 {
1872 let compute_hash = checksum_mode != ContentChecksum::None
1873 && state.frame_header.descriptor.content_checksum_flag();
1874 state.decoder_scratch.set_compute_hash(compute_hash);
1875 }
1876
1877 // Streaming entry point: pre-reserve the backing buffer to
1878 // the FCS-capped window plus a block so multi-block frames don't pay
1879 // repeated `reserve_amortized` grow steps (128 KiB → 256 KiB → ... →
1880 // window) as blocks accumulate, nor a copy of the window when it
1881 // fills. `decode_all` does the same up front in `decode_all_impl`;
1882 // this mirrors it for callers driving `decode_blocks` directly.
1883 // Idempotent — the backend's `reserve` early-returns when capacity
1884 // is already sufficient.
1885 state.reserve_decoding_buffer();
1886
1887 let mut block_dec = decoding::block_decoder::with_kernel(self.kernel);
1888
1889 let buffer_size_before = state.decoder_scratch.buffer_len();
1890 let block_counter_before = state.block_counter;
1891 loop {
1892 vprintln!("################");
1893 vprintln!("Next Block: {}", state.block_counter);
1894 vprintln!("################");
1895 // Capture the failing-block coordinates BEFORE the header read so
1896 // the error carries where it happened: `bytes_read_counter` is the
1897 // frame-absolute offset of this block's header (not yet advanced),
1898 // `block_counter` its 0-based index. Used by both the header- and
1899 // body-error builders below (block-precise recovery under `lsm`).
1900 let block_index = state.block_counter as u32;
1901 let block_frame_offset = state.bytes_read_counter as u32;
1902 let (block_header, block_header_size) =
1903 block_dec.read_block_header(&mut source).map_err(|source| {
1904 block_header_decode_error(source, block_index, block_frame_offset)
1905 })?;
1906 state.bytes_read_counter += u64::from(block_header_size);
1907
1908 vprintln!();
1909 vprintln!(
1910 "Found {} block with size: {}, which will be of size: {}",
1911 block_header.block_type,
1912 block_header.content_size,
1913 block_header.decompressed_size
1914 );
1915
1916 #[cfg(all(feature = "lsm", feature = "hash"))]
1917 let len_before_block: Option<usize> = if self.per_block_checksums_enabled {
1918 Some(state.decoder_scratch.buffer_len())
1919 } else {
1920 None
1921 };
1922 // Only expose the held dictionary while THIS frame is dict-backed
1923 // (`using_dict` is set per dict-apply, cleared on reset). A reused
1924 // decoder keeps `active_dict` across a no-dict frame for the
1925 // `ptr::eq` reuse-skip, so it must be gated here or a stray
1926 // out-of-window offset on a dictless frame would resolve against the
1927 // stale dictionary content instead of erroring.
1928 let dict_ref = if state.using_dict.is_some() {
1929 state.active_dict.as_ref().map(|h| h.as_dict())
1930 } else {
1931 None
1932 };
1933 let bytes_read_in_block_body = state
1934 .decoder_scratch
1935 .decode_block_content(&mut block_dec, &block_header, &mut source, dict_ref)
1936 .map_err(|source| {
1937 block_body_decode_error(
1938 source,
1939 block_index,
1940 block_frame_offset,
1941 &block_header,
1942 block_header_size,
1943 )
1944 })?;
1945 state.bytes_read_counter += bytes_read_in_block_body;
1946
1947 // Per-block XXH64 (low 32 bits) of the just-decompressed
1948 // bytes. Hashed from `last_n_as_slices` so RingBuffer wrap
1949 // is handled in-place, no extra copy.
1950 #[cfg(all(feature = "lsm", feature = "hash"))]
1951 if let Some(len_before_block) = len_before_block {
1952 let added = state.decoder_scratch.buffer_len() - len_before_block;
1953 let (s1, s2) = state.decoder_scratch.last_n_as_slices(added);
1954 let mut h = twox_hash::XxHash64::with_seed(0);
1955 use core::hash::Hasher;
1956 h.write(s1);
1957 h.write(s2);
1958 self.computed_block_checksums.push(h.finish() as u32);
1959 }
1960
1961 state.block_counter += 1;
1962
1963 vprintln!("Output: {}", state.decoder_scratch.buffer_len());
1964
1965 if block_header.last_block {
1966 state.frame_finished = true;
1967 if state.frame_header.descriptor.content_checksum_flag() {
1968 let mut chksum = [0u8; 4];
1969 source
1970 .read_exact(&mut chksum)
1971 .map_err(err::FailedToReadChecksum)?;
1972 state.bytes_read_counter += 4;
1973 let chksum = u32::from_le_bytes(chksum);
1974 state.check_sum = Some(chksum);
1975 }
1976 break;
1977 }
1978
1979 match strat {
1980 BlockDecodingStrategy::All => { /* keep going */ }
1981 BlockDecodingStrategy::UptoBlocks(n) => {
1982 if state.block_counter - block_counter_before >= n {
1983 break;
1984 }
1985 }
1986 BlockDecodingStrategy::UptoBytes(n) => {
1987 if state.decoder_scratch.buffer_len() - buffer_size_before >= n {
1988 break;
1989 }
1990 }
1991 }
1992 }
1993
1994 Ok(state.frame_finished)
1995 }
1996
1997 /// Decode the inner blocks `[start_block, end_block)` of the current
1998 /// frame and return their decompressed bytes as one contiguous buffer.
1999 ///
2000 /// Serves two consumer needs with one call:
2001 ///
2002 /// - **Range-query performance:** decode only the inner zstd blocks that
2003 /// cover a key range instead of the whole frame. Blocks before
2004 /// `start_block` are decoded into the window (zstd blocks share one
2005 /// window, so a leading block's bytes may be the match source for an
2006 /// in-range block and cannot simply be skipped) but their output is not
2007 /// returned; blocks at or after `end_block` are not decoded at all,
2008 /// which is the trailing-block work saving. Map a decompressed byte
2009 /// offset to a block index with
2010 /// [`FrameEmitInfo::decompressed_byte_range`].
2011 /// - **Best-effort recovery:** if a block decode fails, decoding stops,
2012 /// the clean prefix of in-range output is preserved in
2013 /// [`PartialDecode::data`], and the failure is reported via
2014 /// [`PartialDecode::stopped_at`]. Passing `(0, u32::MAX)` decodes the
2015 /// whole frame, stopping at the first corrupt block (pure recovery).
2016 ///
2017 /// `end_block` is exclusive; pass `u32::MAX` to decode to the end of the
2018 /// frame. Call on a freshly [`reset`](Self::reset) decoder (it decodes
2019 /// from the frame's first block).
2020 ///
2021 /// # Resume (cold incremental / top-up)
2022 ///
2023 /// A plain call drains its in-range output from the match window on return,
2024 /// so two consecutive calls cannot resume one another and growing a decoded
2025 /// extent would mean re-decoding the covering prefix from block 0
2026 /// (`O(extent)` per growth, `O(N²)` for a forward walk). The `resume` /
2027 /// `emit_resume` arguments make a symmetric one-call grow-loop possible:
2028 ///
2029 /// - `emit_resume = true` captures the cross-block carry-over state (entropy
2030 /// tables + repcode history + the next block index / output offset) into
2031 /// [`PartialDecode::resume_state`]. The entropy-table snapshot clone is
2032 /// only paid when this is set. The snapshot is `None` when the decode
2033 /// reaches the frame's last block ([`PartialDecode::frame_finished`]):
2034 /// there is no following block to resume from, so an incremental walk
2035 /// stops on `frame_finished` rather than on a `None` snapshot.
2036 /// - `resume = Some(`[`ResumeInput`]`)` continues from a previously emitted
2037 /// [`ResumeState`] WITHOUT re-decompressing the preceding blocks: the
2038 /// match window is primed from [`ResumeInput::window_prime`] and the
2039 /// entropy/repcode tables are restored from the state, so a `Repeat_Mode`
2040 /// resume block resolves byte-identically to a contiguous decode — even
2041 /// across a dropped (cold) decoder.
2042 ///
2043 /// When `resume` is `Some`, decoding resumes at
2044 /// [`ResumeState::block_index`] and the `start_block` argument is ignored
2045 /// (pass `resume.state.block_index()`); position `source` at that block's
2046 /// compressed frame offset
2047 /// ([`FrameEmitInfo::blocks`]`[block_index].offset_in_frame`). After a
2048 /// resumed call, [`bytes_read_from_source`](Self::bytes_read_from_source)
2049 /// and any `stopped_at` offsets are relative to the repositioned `source`.
2050 ///
2051 /// **Dictionaries:** [`ResumeState`] does NOT carry the dictionary content.
2052 /// For a dictionary frame, attach the dictionary to the resuming decoder the
2053 /// same way as for a fresh decode — [`reset`](Self::reset) with the
2054 /// dictionary registered (or
2055 /// [`reset_with_dict_handle`](Self::reset_with_dict_handle)) BEFORE this
2056 /// call — so dict-sourced matches near the frame start resolve. The caller
2057 /// already holds the dictionary (it supplied it at encode time), so
2058 /// re-supplying it on resume is free; storing it in the snapshot would only
2059 /// duplicate it. The resume guard records the applied dictionary's identity
2060 /// and rejects ([`FrameDecoderError::ResumeFrameMismatch`]) a resume whose
2061 /// active dictionary differs from the one the snapshot was captured under.
2062 ///
2063 /// # Errors
2064 ///
2065 /// Returns [`FrameDecoderError::NotYetInitialized`] if the decoder has not
2066 /// been reset, [`FrameDecoderError::InvalidBlockRange`] if the effective
2067 /// start exceeds `end_block`, [`FrameDecoderError::ResumeWindowTooShort`]
2068 /// if `resume`'s `window_prime` is shorter than the match window the resume
2069 /// block can reach back into (`min(window_size, output_offset)`), and
2070 /// [`FrameDecoderError::ResumeFrameMismatch`] if the snapshot was captured
2071 /// from a frame with a different decode shape / dictionary, or (with the
2072 /// `hash` feature) a `window_prime` whose content does not match what was
2073 /// captured — all rejected up front rather than silently mis-resolving
2074 /// matches. A corrupt block is NOT an `Err` here: it is reported via
2075 /// [`PartialDecode::stopped_at`] so the clean prefix survives.
2076 ///
2077 /// [`FrameEmitInfo::decompressed_byte_range`]: crate::encoding::frame_emit_info::FrameEmitInfo::decompressed_byte_range
2078 /// [`FrameEmitInfo::blocks`]: crate::encoding::frame_emit_info::FrameEmitInfo::blocks
2079 #[cfg(feature = "lsm")]
2080 #[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
2081 pub fn decode_blocks_partial(
2082 &mut self,
2083 mut source: impl Read,
2084 start_block: u32,
2085 end_block: u32,
2086 resume: Option<ResumeInput<'_>>,
2087 emit_resume: bool,
2088 ) -> Result<PartialDecode, FrameDecoderError> {
2089 use FrameDecoderError as err;
2090 #[cfg(feature = "hash")]
2091 let checksum_mode = self.content_checksum;
2092 let magicless = self.magicless;
2093 let state = self.state.as_mut().ok_or(err::NotYetInitialized)?;
2094
2095 // Honor the checksum mode before any drain/read can hash: `None` must
2096 // compute no XXH64. `decode_blocks` sets this; the partial path must too,
2097 // or a reused scratch keeps hashing with the default-enabled state.
2098 #[cfg(feature = "hash")]
2099 {
2100 let compute_hash = checksum_mode != ContentChecksum::None
2101 && state.frame_header.descriptor.content_checksum_flag();
2102 state.decoder_scratch.set_compute_hash(compute_hash);
2103 }
2104
2105 // A dictionary that carries no ID cannot be told from another one, and
2106 // the frame key records only that ID: two different raw-content
2107 // dictionaries key alike, so a snapshot would be restored under the
2108 // wrong one — and one taken here could later be. Both directions are
2109 // refused, and refused HERE, before a block is read or a buffer
2110 // reserved. An answer that does not depend on the blocks must not be
2111 // given after decoding them: it would discard the output they produced
2112 // and leave the source and the decoder somewhere the caller cannot
2113 // retry from.
2114 if (resume.is_some() || emit_resume) && state.using_dict == Some(0) {
2115 return Err(err::ResumeUnidentifiedDictionary);
2116 }
2117
2118 // Mirror `decode_blocks`: pre-reserve the backing buffer to the
2119 // FCS-capped window plus a block so multi-block frames don't pay
2120 // repeated grow steps. The RAW frame window stays separately bound —
2121 // the resume logic below bounds match reach by the frame's window
2122 // semantics, not by the (possibly smaller) reservation cap.
2123 let window_size = state.frame_header.window_size().unwrap_or(0) as usize;
2124 state.reserve_decoding_buffer();
2125
2126 // Cold resume: prime the match window + restore entropy/repcode state +
2127 // advance the block cursor BEFORE the loop, so the first in-range block
2128 // resolves its matches and `Repeat_Mode` tables against the caller's
2129 // persisted state instead of re-decoded prefix blocks. The effective
2130 // start is the resume state's block index (the passed `start_block` is
2131 // ignored in resume mode, per the doc).
2132 let effective_start = if let Some(r) = resume {
2133 // Reject a snapshot captured from a different frame shape BEFORE
2134 // touching any decoder state: restoring entropy/repcode tables that
2135 // belong to another frame would silently produce byte-wrong output.
2136 let current_key = FrameKey::from_state(state, magicless);
2137 if current_key != r.state.frame_key {
2138 return Err(err::ResumeFrameMismatch);
2139 }
2140 let output_offset = r.state.output_offset;
2141 // The window the resume block can reach back into is bounded by the
2142 // smaller of the frame's window_size and the bytes produced so far.
2143 let required = core::cmp::min(window_size as u64, output_offset) as usize;
2144 if r.window_prime.len() < required {
2145 return Err(err::ResumeWindowTooShort {
2146 got: r.window_prime.len(),
2147 need: required,
2148 });
2149 }
2150 // Only the most recent `window_size` bytes can ever back a match
2151 // (offset <= window_size by the frame invariant); load just those
2152 // even if the caller handed us a longer prefix, bounding resume
2153 // memory to one window regardless of the skipped prefix's size.
2154 let prime = if r.window_prime.len() > window_size {
2155 &r.window_prime[r.window_prime.len() - window_size..]
2156 } else {
2157 r.window_prime
2158 };
2159 // Content-exact identity: the primed window must hash to what was
2160 // captured at emit. Catches a same-shape-but-different-frame
2161 // snapshot and a wrong/corrupted window_prime (which FrameKey alone
2162 // cannot), before any state is restored. O(window) one-time per
2163 // resume — negligible next to the decode it guards.
2164 #[cfg(feature = "hash")]
2165 if xxh64_of(prime) != r.state.window_hash {
2166 return Err(err::ResumeFrameMismatch);
2167 }
2168 // Validate the effective range (resume mode begins at the resume
2169 // block, ignoring the caller's `start_block`) BEFORE mutating the
2170 // decoder: an inverted `end_block` must fail without priming the
2171 // window / entropy or advancing the cursor, leaving the decoder
2172 // re-resettable rather than in a half-resumed state.
2173 let effective_start = r.state.block_index;
2174 if effective_start > end_block {
2175 return Err(err::InvalidBlockRange {
2176 start_block: effective_start,
2177 end_block,
2178 });
2179 }
2180 state.decoder_scratch.restore_entropy(r.state);
2181 state.decoder_scratch.prime_window(prime, output_offset);
2182 state.block_counter = effective_start as usize;
2183 // The caller repositions `source` to the resume block; report
2184 // consumed bytes relative to that point (reset left this at the
2185 // frame-header size).
2186 state.bytes_read_counter = 0;
2187 effective_start
2188 } else {
2189 // Fresh decode: validate the caller's range (no state to mutate).
2190 if start_block > end_block {
2191 return Err(err::InvalidBlockRange {
2192 start_block,
2193 end_block,
2194 });
2195 }
2196 start_block
2197 };
2198
2199 let mut block_dec = decoding::block_decoder::with_kernel(self.kernel);
2200
2201 // Bytes of prefix-window output that physically precede the first
2202 // in-range block in the buffer. Captured at the prefix → in-range
2203 // transition (after leading blocks were dropped to the window) so we
2204 // can discard exactly those bytes once decoding is done. `None` until
2205 // the first in-range block is reached.
2206 let mut prefix_window_len: Option<usize> = None;
2207 // Exact count of clean in-range decompressed bytes (sum of per-block
2208 // length deltas of the in-range blocks that succeeded). Any partial
2209 // bytes of a failing in-range block are excluded — the fused executor
2210 // rolls the buffer back to the pre-block checkpoint on a sequence
2211 // error, and anything left over is never counted here, so it is not
2212 // drained into `data`.
2213 let mut subset_bytes: u64 = 0;
2214 let mut blocks_decoded: u32 = 0;
2215 let mut stopped_at: Option<(u32, FrameDecoderError)> = None;
2216
2217 loop {
2218 let block_index = state.block_counter as u32;
2219 // Stop before decoding `end_block`: the trailing blocks are never
2220 // touched (the perf win), and the frame's tail is left unread.
2221 if block_index >= end_block || state.frame_finished {
2222 break;
2223 }
2224 let in_range = block_index >= effective_start;
2225 // Snapshot the window length at the prefix → in-range boundary.
2226 if in_range && prefix_window_len.is_none() {
2227 prefix_window_len = Some(state.decoder_scratch.buffer_len());
2228 }
2229
2230 let block_frame_offset = state.bytes_read_counter as u32;
2231 let (block_header, block_header_size) = match block_dec.read_block_header(&mut source) {
2232 Ok(v) => v,
2233 Err(e) => {
2234 stopped_at = Some((
2235 block_index,
2236 block_header_decode_error(e, block_index, block_frame_offset),
2237 ));
2238 break;
2239 }
2240 };
2241 state.bytes_read_counter += u64::from(block_header_size);
2242
2243 let len_before = state.decoder_scratch.buffer_len();
2244 // Only expose the held dictionary while THIS frame is dict-backed
2245 // (`using_dict` is set per dict-apply, cleared on reset). A reused
2246 // decoder keeps `active_dict` across a no-dict frame for the
2247 // `ptr::eq` reuse-skip, so it must be gated here or a stray
2248 // out-of-window offset on a dictless frame would resolve against the
2249 // stale dictionary content instead of erroring.
2250 let dict_ref = if state.using_dict.is_some() {
2251 state.active_dict.as_ref().map(|h| h.as_dict())
2252 } else {
2253 None
2254 };
2255 match state.decoder_scratch.decode_block_content(
2256 &mut block_dec,
2257 &block_header,
2258 &mut source,
2259 dict_ref,
2260 ) {
2261 Ok(body_read) => state.bytes_read_counter += body_read,
2262 Err(e) => {
2263 stopped_at = Some((
2264 block_index,
2265 block_body_decode_error(
2266 e,
2267 block_index,
2268 block_frame_offset,
2269 &block_header,
2270 block_header_size,
2271 ),
2272 ));
2273 break;
2274 }
2275 }
2276 let produced = state.decoder_scratch.buffer_len() - len_before;
2277 // Per-block XXH64 capture, mirroring `decode_blocks`: hash this
2278 // block's just-decoded bytes BEFORE any window drop so the digest
2279 // count stays 1:1 with the blocks decoded on this path too. Covers
2280 // context (out-of-range) blocks as well, matching `decode_blocks`
2281 // which hashes every block it decodes.
2282 #[cfg(all(feature = "lsm", feature = "hash"))]
2283 if self.per_block_checksums_enabled {
2284 use core::hash::Hasher;
2285 let (s1, s2) = state.decoder_scratch.last_n_as_slices(produced);
2286 let mut h = twox_hash::XxHash64::with_seed(0);
2287 h.write(s1);
2288 h.write(s2);
2289 self.computed_block_checksums.push(h.finish() as u32);
2290 }
2291 state.block_counter += 1;
2292 if in_range {
2293 subset_bytes += produced as u64;
2294 blocks_decoded += 1;
2295 }
2296
2297 if block_header.last_block {
2298 state.frame_finished = true;
2299 if state.frame_header.descriptor.content_checksum_flag() {
2300 let mut chksum = [0u8; 4];
2301 match source.read_exact(&mut chksum) {
2302 Ok(()) => {
2303 state.bytes_read_counter += 4;
2304 state.check_sum = Some(u32::from_le_bytes(chksum));
2305 }
2306 // A trailing-checksum read failure does not invalidate
2307 // the decoded bytes; surface it so the caller knows the
2308 // frame tail was truncated, but keep `data`.
2309 Err(e) => {
2310 stopped_at = Some((block_index, err::FailedToReadChecksum(e)));
2311 }
2312 }
2313 }
2314 break;
2315 }
2316
2317 // Leading (out-of-range) block: bound memory to the window. We
2318 // must NOT drop once in-range, or the in-range output we are about
2319 // to return would be discarded.
2320 if !in_range {
2321 state.decoder_scratch.buffer_drop_to_window_size();
2322 }
2323 }
2324
2325 // Emit cross-block carry-over state for a later resume, if requested.
2326 // Captured AFTER the loop (entropy tables / repcode history are final)
2327 // but BEFORE the drain — the drain only touches the visible output, not
2328 // the entropy state or `total_output_counter`. `block_counter` /
2329 // `total_output()` give the resume coordinates: the next block to decode
2330 // and the cumulative decompressed offset before it (clean even after an
2331 // early stop, since a failed block rolls both back to its checkpoint).
2332 // Suppress the snapshot on the terminal block: `block_counter` is then
2333 // one past the last block (EOF), for which there is no next-block source
2334 // position to resume from. A resume needs a real following block.
2335 let resume_state = if emit_resume && !state.frame_finished {
2336 let dict_ref = if state.using_dict.is_some() {
2337 state.active_dict.as_ref().map(|h| h.as_dict())
2338 } else {
2339 None
2340 };
2341 let (fse, huf, offset_hist) = state.decoder_scratch.export_entropy(dict_ref);
2342 Some(ResumeState {
2343 frame_key: FrameKey::from_state(state, magicless),
2344 block_index: state.block_counter as u32,
2345 output_offset: state.decoder_scratch.total_output(),
2346 fse,
2347 huf,
2348 offset_hist,
2349 #[cfg(feature = "hash")]
2350 window_hash: state.decoder_scratch.window_tail_hash(window_size),
2351 })
2352 } else {
2353 None
2354 };
2355
2356 // The visible buffer is now `[prefix window][in-range clean][maybe
2357 // trailing garbage from a failed in-range block]`. Drop the prefix
2358 // window from the front (match resolution is complete, so it is no
2359 // longer needed), then drain exactly the clean in-range byte count.
2360 let w = prefix_window_len.unwrap_or(0);
2361 state.decoder_scratch.buffer_discard_front(w);
2362 let mut data = alloc::vec![0u8; subset_bytes as usize];
2363 state
2364 .decoder_scratch
2365 .buffer_read_all(&mut data)
2366 .map_err(err::FailedToDrainDecodebuffer)?;
2367
2368 // Clear anything still buffered so a later `read()`/`collect()` on this
2369 // decoder cannot surface out-of-range bytes: the leading-block window
2370 // when no in-range block was reached (`prefix_window_len` stayed
2371 // `None`, so `w` was 0), or trailing garbage from a failed in-range
2372 // block. Only the returned `data` is the partial decode's output.
2373 let residual = state.decoder_scratch.buffer_len();
2374 state.decoder_scratch.buffer_discard_front(residual);
2375
2376 Ok(PartialDecode {
2377 data,
2378 start_block: effective_start,
2379 blocks_decoded,
2380 stopped_at,
2381 frame_finished: state.frame_finished,
2382 resume_state,
2383 })
2384 }
2385
2386 /// Collect bytes and retain window_size bytes while decoding is still going on.
2387 /// After decoding of the frame (is_finished() == true) has finished it will collect all remaining bytes
2388 pub fn collect(&mut self) -> Option<Vec<u8>> {
2389 let finished = self.is_finished();
2390 let state = self.state.as_mut()?;
2391 if finished {
2392 Some(state.decoder_scratch.buffer_drain())
2393 } else {
2394 state.decoder_scratch.buffer_drain_to_window_size()
2395 }
2396 }
2397
2398 /// Collect bytes and retain window_size bytes while decoding is still going on.
2399 /// After decoding of the frame (is_finished() == true) has finished it will collect all remaining bytes
2400 pub fn collect_to_writer(&mut self, w: impl Write) -> Result<usize, Error> {
2401 let finished = self.is_finished();
2402 let state = match &mut self.state {
2403 None => return Ok(0),
2404 Some(s) => s,
2405 };
2406 if finished {
2407 state.decoder_scratch.buffer_drain_to_writer(w)
2408 } else {
2409 state.decoder_scratch.buffer_drain_to_window_size_writer(w)
2410 }
2411 }
2412
2413 /// How many bytes can currently be collected from the decodebuffer, while decoding is going on this will be lower than the actual decodbuffer size
2414 /// because window_size bytes need to be retained for decoding.
2415 /// After decoding of the frame (is_finished() == true) has finished it will report all remaining bytes
2416 pub fn can_collect(&self) -> usize {
2417 let finished = self.is_finished();
2418 let state = match &self.state {
2419 None => return 0,
2420 Some(s) => s,
2421 };
2422 if finished {
2423 state.decoder_scratch.buffer_can_drain()
2424 } else {
2425 state
2426 .decoder_scratch
2427 .buffer_can_drain_to_window_size()
2428 .unwrap_or(0)
2429 }
2430 }
2431
2432 /// Decodes blocks from the source slice and reads from the decodebuffer into the target slice, one block at a
2433 /// time: output is handed to `target` as each block completes, and no further block is decoded while `target`
2434 /// cannot take what is already decoded. The decode buffer so holds one window plus one block however much input
2435 /// is supplied; call again with the unread input to continue.
2436 /// The source slice may contain only parts of a frame but must contain at least one full block to make progress
2437 ///
2438 /// By all means use decode_blocks if you have a io.Reader available. This is just for compatibility with other decompressors
2439 /// which try to serve an old-style c api
2440 ///
2441 /// Returns (read, written). Both zero means the call made no progress: the
2442 /// source holds no full block and the buffer no drainable output, so the
2443 /// same input cannot advance. `read == 0` with `written > 0` is progress of
2444 /// the other kind: `target` filled from output already buffered, and the
2445 /// same input decodes further once the caller offers more room.
2446 ///
2447 /// Note that no kind of block can be bigger than 128kb.
2448 /// So to be safe use at least 128*1024 (max block content size) + 3 (block_header size) + 18 (max frame_header size) bytes as your source buffer
2449 ///
2450 /// You may call this function with an empty source after all bytes have been decoded. This is equivalent to just call decoder.read(&mut target)
2451 pub fn decode_from_to(
2452 &mut self,
2453 source: &[u8],
2454 target: &mut [u8],
2455 ) -> Result<(usize, usize), FrameDecoderError> {
2456 use FrameDecoderError as err;
2457 let bytes_read_at_start = match &self.state {
2458 Some(s) => s.bytes_read_counter,
2459 None => 0,
2460 };
2461 // Bytes already handed to `target` by the per-block drain below.
2462 let mut written = 0usize;
2463
2464 if !self.is_finished() || self.state.is_none() {
2465 let mut mt_source = source;
2466
2467 if self.state.is_none() {
2468 self.init(&mut mt_source)?;
2469 }
2470
2471 //pseudo block to scope "state" so we can borrow self again after the block
2472 {
2473 let kernel = self.kernel;
2474 let state = match &mut self.state {
2475 Some(s) => s,
2476 None => panic!("Bug in library"),
2477 };
2478 let mut block_dec = decoding::block_decoder::with_kernel(kernel);
2479
2480 // Honour the content-checksum mode on this hand-rolled decode
2481 // loop (it does not go through `decode_blocks`): hash only when
2482 // a digest is wanted and the frame carries one. `None` skips the
2483 // XXH64 pass; verification happens after the final drain below.
2484 #[cfg(feature = "hash")]
2485 {
2486 let compute_hash = self.content_checksum != ContentChecksum::None
2487 && state.frame_header.descriptor.content_checksum_flag();
2488 state.decoder_scratch.set_compute_hash(compute_hash);
2489 }
2490
2491 if state.frame_header.descriptor.content_checksum_flag()
2492 && state.frame_finished
2493 && state.check_sum.is_none()
2494 {
2495 // The trailing checksum arrived on a separate call (the last
2496 // block finished earlier). Consume it and fall through to the
2497 // shared `self.read` + post-drain verify below — NOT an early
2498 // return — so any output still buffered from a prior
2499 // small-`target` call is flushed on this call too, and the
2500 // checksum is verified through the one shared path.
2501 if mt_source.len() >= 4 {
2502 let chksum = mt_source[..4].try_into().expect("optimized away");
2503 state.bytes_read_counter += 4;
2504 let chksum = u32::from_le_bytes(chksum);
2505 state.check_sum = Some(chksum);
2506 mt_source = &mt_source[4..];
2507 }
2508 }
2509
2510 loop {
2511 // The frame is fully decoded (last block seen, trailer
2512 // consumed above); no more blocks to read. Any leftover
2513 // bytes are not a block header — stop before misreading them.
2514 if state.frame_finished {
2515 break;
2516 }
2517 // Hand what the window no longer needs to `target` before
2518 // decoding more, and decode no further while `target` cannot
2519 // take it: the buffer then holds one window plus the block
2520 // being decoded, whatever the caller supplies, as upstream
2521 // `ZSTD_decompressStream` flushes each block before the next.
2522 let (read, pending) = state
2523 .decoder_scratch
2524 .buffer_read_reporting_pending(&mut target[written..])
2525 .map_err(err::FailedToDrainDecodebuffer)?;
2526 written += read;
2527 // Stop on a full target as well as on output left behind:
2528 // a drain that empties the buffer into the last of `target`
2529 // leaves nothing pending, and decoding another block then
2530 // consumes input the caller cannot be handed the output of.
2531 // A target of no bytes is not full in that sense: it starts
2532 // at its own length, so the test would fire before any block
2533 // was read and a frame that produces nothing could never
2534 // reach the empty block that ends it. Such a frame decodes
2535 // here; one that does produce bytes buffers its first block
2536 // and stops on the `pending` arm of the next pass.
2537 if pending > 0 || (!target.is_empty() && written == target.len()) {
2538 break;
2539 }
2540 //check if there are enough bytes for the next header
2541 if mt_source.len() < 3 {
2542 break;
2543 }
2544 let block_index = state.block_counter as u32;
2545 let block_frame_offset = state.bytes_read_counter as u32;
2546 let (block_header, block_header_size) = block_dec
2547 .read_block_header(&mut mt_source)
2548 .map_err(|source| {
2549 block_header_decode_error(source, block_index, block_frame_offset)
2550 })?;
2551
2552 // check the needed size for the block before updating counters.
2553 // If not enough bytes are in the source, the header will have to be read again, so act like we never read it in the first place
2554 if mt_source.len() < block_header.content_size as usize {
2555 break;
2556 }
2557 state.bytes_read_counter += u64::from(block_header_size);
2558 // A frame that declares its size gets its buffer in one
2559 // allocation once its first block is in hand, as upstream
2560 // allocates its stream buffer per frame: growing it block by
2561 // block cost a fresh decoder several reallocations, copies
2562 // and page-fault passes per frame. Not on the header alone,
2563 // which would let a header followed by nothing reserve its
2564 // whole declared window. The size is content-capped, so a
2565 // small frame gets a small buffer; a frame of unknown size
2566 // keeps growing lazily rather than paying for its window.
2567 if state.block_counter == 0 && state.frame_header.fcs_declared() {
2568 state.reserve_decoding_buffer();
2569 }
2570
2571 // Only expose the held dictionary while THIS frame is dict-backed
2572 // (`using_dict` is set per dict-apply, cleared on reset). A reused
2573 // decoder keeps `active_dict` across a no-dict frame for the
2574 // `ptr::eq` reuse-skip, so it must be gated here or a stray
2575 // out-of-window offset on a dictless frame would resolve against the
2576 // stale dictionary content instead of erroring.
2577 let dict_ref = if state.using_dict.is_some() {
2578 state.active_dict.as_ref().map(|h| h.as_dict())
2579 } else {
2580 None
2581 };
2582 let bytes_read_in_block_body = state
2583 .decoder_scratch
2584 .decode_block_content(
2585 &mut block_dec,
2586 &block_header,
2587 &mut mt_source,
2588 dict_ref,
2589 )
2590 .map_err(|source| {
2591 block_body_decode_error(
2592 source,
2593 block_index,
2594 block_frame_offset,
2595 &block_header,
2596 block_header_size,
2597 )
2598 })?;
2599 state.bytes_read_counter += bytes_read_in_block_body;
2600 state.block_counter += 1;
2601
2602 if block_header.last_block {
2603 state.frame_finished = true;
2604 if state.frame_header.descriptor.content_checksum_flag() {
2605 //if there are enough bytes handle this here. Else the block at the start of this function will handle it at the next call
2606 if mt_source.len() >= 4 {
2607 let chksum = mt_source[..4].try_into().expect("optimized away");
2608 state.bytes_read_counter += 4;
2609 let chksum = u32::from_le_bytes(chksum);
2610 state.check_sum = Some(chksum);
2611 }
2612 }
2613 break;
2614 }
2615 }
2616 }
2617 }
2618
2619 let result_len = written
2620 + self
2621 .read(&mut target[written..])
2622 .map_err(err::FailedToDrainDecodebuffer)?;
2623 // Once the frame is fully decoded and drained, the running digest is
2624 // final: validate it in `Verify` mode (no-op otherwise). Same finish
2625 // point as the streaming reader.
2626 #[cfg(feature = "hash")]
2627 if self.is_finished() && self.can_collect() == 0 {
2628 self.verify_content_checksum()?;
2629 }
2630 let bytes_read_at_end = match &mut self.state {
2631 Some(s) => s.bytes_read_counter,
2632 None => panic!("Bug in library"),
2633 };
2634 let read_len = bytes_read_at_end - bytes_read_at_start;
2635 Ok((read_len as usize, result_len))
2636 }
2637
2638 /// Decode multiple frames into the output slice.
2639 ///
2640 /// `input` must contain an exact number of frames. Skippable frames are allowed and will be
2641 /// skipped during decode.
2642 ///
2643 /// `output` must be large enough to hold the decompressed data. If you don't know
2644 /// how large the output will be, use [`FrameDecoder::decode_blocks`] instead.
2645 ///
2646 /// This calls [`FrameDecoder::init`], and all bytes currently in the decoder will be lost.
2647 ///
2648 /// Returns the number of bytes written to `output`.
2649 pub fn decode_all(
2650 &mut self,
2651 input: &[u8],
2652 output: &mut [u8],
2653 ) -> Result<usize, FrameDecoderError> {
2654 #[cfg(not(feature = "lsm"))]
2655 {
2656 self.decode_all_impl(input, output, |this, src| this.reset_from_slice(src))
2657 }
2658 #[cfg(feature = "lsm")]
2659 {
2660 self.decode_all_impl(input, output, |this, src| this.reset_from_slice(src), None)
2661 }
2662 }
2663
2664 /// Decode multiple frames into the output slice, invoking `visitor`
2665 /// for every skippable frame encountered before advancing past it.
2666 ///
2667 /// `input` must contain an exact number of frames. Skippable frames
2668 /// (RFC 8878 §3.1.2 magic numbers `0x184D2A50..=0x184D2A5F`) are
2669 /// allowed and will be both visited AND skipped: the visitor gets
2670 /// `(magic_variant, payload)` where `magic_variant` is the low
2671 /// nibble of the magic (`magic - 0x184D2A50`, range `0..=15`) and
2672 /// `payload` is a borrowed slice of the on-wire payload bytes (the
2673 /// skippable frame's `Frame_Size` field worth of data) into
2674 /// `input` — no allocation.
2675 ///
2676 /// The visitor sees skippable frames in stream order; interleaved
2677 /// regular zstd frames continue to decompress into `output` exactly
2678 /// as `decode_all` does.
2679 ///
2680 /// `output` must be large enough to hold the decompressed data.
2681 /// Returns the number of bytes written to `output`.
2682 ///
2683 /// # Example
2684 ///
2685 /// ```ignore
2686 /// use structured_zstd::decoding::FrameDecoder;
2687 ///
2688 /// let mut decoder = FrameDecoder::new();
2689 /// let mut output = vec![0u8; 1024];
2690 /// let mut collected: Vec<(u8, Vec<u8>)> = Vec::new();
2691 /// let n = decoder.decode_all_with_skippable_visitor(
2692 /// input,
2693 /// &mut output,
2694 /// |variant, payload| collected.push((variant, payload.to_vec())),
2695 /// )?;
2696 /// ```
2697 #[cfg(feature = "lsm")]
2698 #[cfg_attr(docsrs, doc(cfg(feature = "lsm")))]
2699 pub fn decode_all_with_skippable_visitor<F>(
2700 &mut self,
2701 input: &[u8],
2702 output: &mut [u8],
2703 mut visitor: F,
2704 ) -> Result<usize, FrameDecoderError>
2705 where
2706 F: FnMut(u8, &[u8]),
2707 {
2708 self.decode_all_impl(
2709 input,
2710 output,
2711 |this, src| this.reset_from_slice(src),
2712 Some(&mut visitor),
2713 )
2714 }
2715
2716 /// Decode multiple frames into the output slice using a pre-parsed dictionary handle.
2717 ///
2718 /// `input` must contain an exact number of frames. Skippable frames are allowed and will be
2719 /// skipped during decode.
2720 ///
2721 /// `output` must be large enough to hold the decompressed data. If you don't know
2722 /// how large the output will be, use [`FrameDecoder::decode_blocks`] instead.
2723 ///
2724 /// This calls [`FrameDecoder::init_with_dict_handle`], and all bytes currently in the
2725 /// decoder will be lost.
2726 ///
2727 /// # Warning
2728 ///
2729 /// Each decoded frame is initialized with `dict`, even when a frame header
2730 /// omits the optional dictionary ID. Callers must only use this API when
2731 /// they already know the input frames were encoded with the provided
2732 /// dictionary; otherwise decoded output can be silently corrupted.
2733 pub fn decode_all_with_dict_handle(
2734 &mut self,
2735 input: &[u8],
2736 output: &mut [u8],
2737 dict: &DictionaryHandle,
2738 ) -> Result<usize, FrameDecoderError> {
2739 #[cfg(not(feature = "lsm"))]
2740 {
2741 self.decode_all_impl(input, output, |this, src| {
2742 this.reset_from_slice_with_dict_handle(src, dict)
2743 })
2744 }
2745 #[cfg(feature = "lsm")]
2746 {
2747 self.decode_all_impl(
2748 input,
2749 output,
2750 |this, src| this.reset_from_slice_with_dict_handle(src, dict),
2751 None,
2752 )
2753 }
2754 }
2755
2756 /// Whether the decoder sits at the very start of an initialised frame:
2757 /// the header has been read (state populated) but no block has been
2758 /// decoded and the frame is not finished. In this state the wrapped
2759 /// source is positioned exactly after the frame header, so
2760 /// [`Self::decode_current_frame_to_vec`] can decode the rest of the frame
2761 /// straight from the remaining source bytes.
2762 pub(crate) fn is_at_frame_start(&self) -> bool {
2763 self.state
2764 .as_ref()
2765 .is_some_and(|s| s.block_counter == 0 && !s.frame_finished)
2766 }
2767
2768 /// Decode the CURRENT (already-initialised) frame, APPENDING the
2769 /// decompressed bytes to `output`, and return the number appended.
2770 ///
2771 /// `input` must be the frame's post-header bytes (the wrapped source after
2772 /// `init` consumed the header). Unlike [`Self::decode_all_to_vec`] this
2773 /// neither re-reads a header nor requires the caller to pre-reserve
2774 /// capacity: a frame that declares its content size decodes DIRECTLY into
2775 /// freshly-grown `output` capacity via the single-copy direct path
2776 /// ([`Self::run_direct_decode`]) — bypassing the `Ring`/`FlatBuf` →
2777 /// `read()` drain copy the streaming loop pays — while an unsized frame
2778 /// falls back to the window-bounded ring drain (still one copy, into
2779 /// `output`). Backs [`StreamingDecoder`](crate::decoding::StreamingDecoder)'s
2780 /// `read_to_end` fast path; the caller must ensure
2781 /// [`Self::is_at_frame_start`].
2782 ///
2783 /// # Errors
2784 ///
2785 /// Propagates any [`FrameDecoderError`] from block decode, content-size
2786 /// mismatch, or (in `Verify` mode) checksum validation.
2787 pub(crate) fn decode_current_frame_to_vec(
2788 &mut self,
2789 mut input: &[u8],
2790 output: &mut Vec<u8>,
2791 keep_dictionary: bool,
2792 ) -> Result<usize, FrameDecoderError> {
2793 let start_len = output.len();
2794 // The current frame is already initialised (its header consumed by the
2795 // caller, WITH the dictionary applied if the decoder was constructed
2796 // with one). Decode it, then decode any FOLLOWING concatenated /
2797 // skippable frames in `input` so the whole source is consumed to EOF
2798 // and nothing is dropped (matching `read_to_end` semantics).
2799 self.decode_one_frame_to_vec(&mut input, output)?;
2800 self.decode_concatenated_frames_to_vec(&mut input, output, keep_dictionary)?;
2801 Ok(output.len() - start_len)
2802 }
2803
2804 /// Initialise and decode every frame remaining in `input` (concatenated /
2805 /// skippable), APPENDING to `output`. `input` is advanced as frames are
2806 /// consumed; on return it is empty. With `keep_dictionary` each following
2807 /// frame is initialised with the dictionary the decoder holds
2808 /// ([`Self::reset_with_active_dict`]), so a forced dictionary is preserved
2809 /// even for frames that omit the dictionary id (plain [`Self::init`] would
2810 /// resolve dictionaries by id only). Backs the `read_to_end` fast path (the
2811 /// frames after the current one) and its mid-frame fallback (the frames
2812 /// after the partially-read one).
2813 pub(crate) fn decode_concatenated_frames_to_vec(
2814 &mut self,
2815 input: &mut &[u8],
2816 output: &mut Vec<u8>,
2817 keep_dictionary: bool,
2818 ) -> Result<usize, FrameDecoderError> {
2819 let start_len = output.len();
2820 while !input.is_empty() {
2821 let init_result = if keep_dictionary {
2822 self.reset_with_active_dict(&mut *input)
2823 } else {
2824 self.init(&mut *input)
2825 };
2826 match init_result {
2827 Ok(_) => {}
2828 Err(FrameDecoderError::ReadFrameHeaderError(
2829 crate::decoding::errors::ReadFrameHeaderError::SkipFrame { length, .. },
2830 )) => {
2831 *input = input
2832 .get(length as usize..)
2833 .ok_or(FrameDecoderError::FailedToSkipFrame)?;
2834 continue;
2835 }
2836 Err(e) => return Err(e),
2837 }
2838 self.decode_one_frame_to_vec(&mut *input, output)?;
2839 }
2840 Ok(output.len() - start_len)
2841 }
2842
2843 /// Decode the single CURRENT (already-initialised) frame, APPENDING to
2844 /// `output`. Helper for [`Self::decode_current_frame_to_vec`].
2845 fn decode_one_frame_to_vec(
2846 &mut self,
2847 input: &mut &[u8],
2848 output: &mut Vec<u8>,
2849 ) -> Result<usize, FrameDecoderError> {
2850 let frame_start = output.len();
2851 let (content_size, fcs_declared) = {
2852 let s = self.state.as_ref().expect("frame is initialised");
2853 (
2854 s.frame_header.frame_content_size(),
2855 s.frame_header.fcs_declared(),
2856 )
2857 };
2858 // Direct path: a declared, non-empty content size that FITS in `usize`
2859 // (and whose end offset does not overflow). `usize::try_from` guards the
2860 // 32-bit / oversized-FCS truncation; an unrepresentable size falls
2861 // through to the window-bounded ring drain rather than allocating a
2862 // truncated buffer that would violate `run_direct_decode`'s precondition.
2863 //
2864 // Plausibility gate: the direct path `resize`s `output` to the declared
2865 // size up front, so a tiny/truncated frame declaring a huge (but
2866 // representable) FCS would allocate + zero that whole size before the
2867 // body is validated. zstd's per-block ceiling is MAX_BLOCK_SIZE from as
2868 // little as ~4 input bytes, so the declared size cannot legitimately
2869 // exceed `input.len() * (MAX_BLOCK_SIZE / 4)`. Anything larger falls
2870 // through to the ring drain, which grows only as real bytes are produced
2871 // and errors out cheaply on truncated input. `input` spans the remaining
2872 // source (this frame plus any following ones), so the bound only ever
2873 // over-permits — a legitimate frame is never forced off the direct path.
2874 // saturating_mul is intentional: an overflow means the available input
2875 // is so large that any representable FCS is plausible (cap = "no limit").
2876 const MAX_DECOMPRESSION_RATIO: usize = (crate::common::MAX_BLOCK_SIZE / 4) as usize;
2877 if content_size > 0
2878 && let Ok(cs) = usize::try_from(content_size)
2879 && cs <= input.len().saturating_mul(MAX_DECOMPRESSION_RATIO)
2880 && let Some(frame_end) = frame_start.checked_add(cs)
2881 {
2882 // Reserve exactly the frame's content and decode straight into it
2883 // (single copy, no ring). The direct path writes precisely
2884 // `content_size` bytes (erroring otherwise), so the grown region is
2885 // fully written.
2886 output.resize(frame_end, 0);
2887 // On error, drop the just-grown (zeroed) tail before propagating so
2888 // callers never observe bytes that were never decoded.
2889 let written = match self.run_direct_decode(
2890 &mut *input,
2891 &mut output[frame_start..],
2892 Some(content_size),
2893 ) {
2894 Ok(n) => n,
2895 Err(e) => {
2896 output.truncate(frame_start);
2897 return Err(e);
2898 }
2899 };
2900 output.truncate(frame_start + written);
2901 #[cfg(feature = "hash")]
2902 self.verify_content_checksum()?;
2903 return Ok(written);
2904 }
2905 // The ring-drain fallback below pre-reserves `useful_window_size()`
2906 // (= `window.min(FCS)`), which for a single-segment frame is the
2907 // declared FCS itself — so a truncated single-segment frame lying about
2908 // its size would still allocate the pledged window before the body
2909 // errors, sidestepping the direct-path gate above. Reject such a frame
2910 // up front when its declared (FCS-bearing) window exceeds what the
2911 // available input could plausibly produce. Frames without a declared
2912 // size keep their window-descriptor reservation (already capped at
2913 // `MAXIMUM_ALLOWED_WINDOW_SIZE` at init); a small-window multi-segment
2914 // frame still falls through to the ring drain, which errors cheaply on
2915 // the truncated body.
2916 if fcs_declared
2917 && let Some(state) = self.state.as_ref()
2918 && state.useful_window_size() > input.len().saturating_mul(MAX_DECOMPRESSION_RATIO)
2919 {
2920 return Err(FrameDecoderError::FrameContentSizeMismatch {
2921 declared: content_size,
2922 produced: 0,
2923 });
2924 }
2925 // No declared size, explicit FCS=0, or an unrepresentable FCS: window-
2926 // bounded ring drain, appended directly to `output` via
2927 // `collect_to_writer` (no staging buffer).
2928 loop {
2929 self.decode_blocks(&mut *input, BlockDecodingStrategy::UptoBytes(1024 * 1024))?;
2930 self.collect_to_writer(&mut *output)
2931 .map_err(FrameDecoderError::FailedToDrainDecodebuffer)?;
2932 if self.is_finished() {
2933 // Final flush of the retained window tail.
2934 self.collect_to_writer(&mut *output)
2935 .map_err(FrameDecoderError::FailedToDrainDecodebuffer)?;
2936 break;
2937 }
2938 }
2939 let produced = (output.len() - frame_start) as u64;
2940 // A declared content size MUST match what the body produced — otherwise
2941 // accept the same corrupt frames `decode_all_impl` rejects (e.g. an
2942 // explicit FCS=0 whose body emits bytes). Use `fcs_declared()` so an
2943 // on-wire FCS=0 is validated, while an unknown size is not.
2944 if fcs_declared && produced != content_size {
2945 return Err(FrameDecoderError::FrameContentSizeMismatch {
2946 declared: content_size,
2947 produced,
2948 });
2949 }
2950 #[cfg(feature = "hash")]
2951 self.verify_content_checksum()?;
2952 Ok(produced as usize)
2953 }
2954
2955 /// Default-feature decode_all_impl: no visitor parameter so the
2956 /// no-lsm build's call surface and codegen are byte-identical to
2957 /// the pre-#172 implementation. Compiles only when `lsm` is OFF.
2958 #[cfg(not(feature = "lsm"))]
2959 fn decode_all_impl(
2960 &mut self,
2961 mut input: &[u8],
2962 mut output: &mut [u8],
2963 mut init_frame: impl FnMut(&mut Self, &mut &[u8]) -> Result<(), FrameDecoderError>,
2964 ) -> Result<usize, FrameDecoderError> {
2965 let mut total_bytes_written = 0;
2966 while !input.is_empty() {
2967 match init_frame(self, &mut input) {
2968 Ok(_) => {}
2969 Err(FrameDecoderError::ReadFrameHeaderError(
2970 crate::decoding::errors::ReadFrameHeaderError::SkipFrame { length, .. },
2971 )) => {
2972 input = input
2973 .get(length as usize..)
2974 .ok_or(FrameDecoderError::FailedToSkipFrame)?;
2975 continue;
2976 }
2977 Err(e) => return Err(e),
2978 };
2979 // Per-frame direct-path dispatch. Now safe to route the
2980 // public `decode_all` here because
2981 // `UserSliceBackend::exec_sequence_inline` returns
2982 // `Result<(), ExecuteSequencesError>` instead of
2983 // panicking on capacity overflow; the error propagates
2984 // up as `FrameDecoderError`. Eligibility (FCS > 0,
2985 // remaining `output` slice holds the declared content)
2986 // puts the frame on the fast path that bypasses the
2987 // FlatBuf/Ring -> `read()` drain copy. Ineligible frames
2988 // (no FCS, output too small) fall through to the legacy
2989 // `decode_blocks` + `read` drain loop below. Dictionary
2990 // frames are eligible: `run_direct_decode` hands the
2991 // shared dict handle to its buffer, and beyond-prefix
2992 // offsets resolve through `repeat_from_dict`.
2993 let (content_size, fcs_declared) = {
2994 let state_ref = self.state.as_ref().expect("init populated state");
2995 (
2996 state_ref.frame_header.frame_content_size(),
2997 state_ref.frame_header.fcs_declared(),
2998 )
2999 };
3000 // Direct decode requires only that the caller slice holds the
3001 // declared content; the inline sequence-exec path no longer
3002 // needs `WILDCOPY_OVERLENGTH` trailing slack because the
3003 // trailing sequence(s) take the bounded (non-overshooting)
3004 // copy in `UserSliceBackend::exec_sequence_bounded`. This is
3005 // the universal "decode into an FCS-sized buffer" case (a
3006 // caller sizing `output` to exactly `frame_content_size`),
3007 // so dropping the slack requirement halves its peak alloc.
3008 //
3009 // Per-block checksums collected inside `run_direct_decode`
3010 // post-loop (over recorded (start, end) ranges of `output`)
3011 // so the direct path stays eligible AND keeps the
3012 // window-size cap (`drop_to_window_size`) between blocks
3013 // that the spec relies on for `offset <= window_size`
3014 // validation. Path choice no longer alters checksum
3015 // semantics.
3016 // A frame that declares no size decodes straight into the
3017 // caller's slice too, the slice being its limit, as upstream
3018 // `ZSTD_decompressDCtx` decodes into `dst`. The drain path
3019 // reserved the frame's whole declared window for it, which for a
3020 // streamed producer's small frame is megabytes for kilobytes.
3021 let declared_size = fcs_declared.then_some(content_size);
3022 let direct_eligible = match declared_size {
3023 Some(declared) => declared > 0 && (output.len() as u64) >= declared,
3024 None => true,
3025 };
3026 if direct_eligible {
3027 let written = self.run_direct_decode(&mut input, output, declared_size)?;
3028 output = &mut output[written..];
3029 total_bytes_written += written;
3030 // Per-frame content-checksum verification (no-op unless the
3031 // mode is `Verify` and the frame carries a checksum).
3032 #[cfg(feature = "hash")]
3033 self.verify_content_checksum()?;
3034 continue;
3035 }
3036 // Non-direct fallback: pre-reserve the backing buffer to
3037 // `window_size` in a single allocation before block decode
3038 // starts, so multi-segment frames don't pay repeated
3039 // `reserve_amortized` grow steps as blocks accumulate (each
3040 // block only reserves MAX_BLOCK_SIZE = 128 KiB, so a window
3041 // > 128 KiB otherwise grows through several intermediate
3042 // sizes with `alloc_zeroed + memcpy` each time).
3043 if let Some(state) = self.state.as_mut() {
3044 // FCS-capped via `decoding_buffer_size` — the same cap
3045 // `decode_blocks` applies, so its per-iteration reserve in
3046 // the loop below cannot grow the buffer back to the raw
3047 // frame window.
3048 state.reserve_decoding_buffer();
3049 }
3050 let frame_start_total = total_bytes_written;
3051 loop {
3052 self.decode_blocks(&mut input, BlockDecodingStrategy::UptoBytes(1024 * 1024))?;
3053 let bytes_written = self
3054 .read(output)
3055 .map_err(FrameDecoderError::FailedToDrainDecodebuffer)?;
3056 output = &mut output[bytes_written..];
3057 total_bytes_written += bytes_written;
3058 if self.can_collect() != 0 {
3059 return Err(FrameDecoderError::TargetTooSmall);
3060 }
3061 if self.is_finished() {
3062 break;
3063 }
3064 }
3065 // Per-frame FCS validation on the legacy fallback path.
3066 // Use `fcs_declared()` (NOT `content_size > 0`) so an
3067 // empty frame with explicit FCS=0 on the wire still gets
3068 // validated.
3069 if fcs_declared {
3070 let produced = (total_bytes_written - frame_start_total) as u64;
3071 if produced != content_size {
3072 return Err(FrameDecoderError::FrameContentSizeMismatch {
3073 declared: content_size,
3074 produced,
3075 });
3076 }
3077 }
3078 // Per-frame content-checksum verification on the drain path: the
3079 // frame is fully decoded and drained here (is_finished + nothing
3080 // left to collect), so the running digest and stored value are
3081 // final. No-op unless the mode is `Verify`.
3082 #[cfg(feature = "hash")]
3083 self.verify_content_checksum()?;
3084 }
3085
3086 Ok(total_bytes_written)
3087 }
3088
3089 /// `lsm`-feature decode_all_impl: adds the optional skippable
3090 /// visitor parameter consumed by
3091 /// [`Self::decode_all_with_skippable_visitor`]. Mirrors the no-lsm
3092 /// variant including the direct-path dispatch + FCS-validation
3093 /// rationale comments, so the two functions stay in sync; the only
3094 /// behavioral difference is the SkipFrame arm, which uses
3095 /// `split_at(length)` (single bounds check) instead of two
3096 /// separate `get(..length)` / `get(length..)` slices and invokes
3097 /// the visitor (when `Some`) on the borrowed payload before
3098 /// advancing past it.
3099 #[cfg(feature = "lsm")]
3100 #[allow(clippy::type_complexity)]
3101 fn decode_all_impl(
3102 &mut self,
3103 mut input: &[u8],
3104 mut output: &mut [u8],
3105 mut init_frame: impl FnMut(&mut Self, &mut &[u8]) -> Result<(), FrameDecoderError>,
3106 mut skippable_visitor: Option<&mut dyn FnMut(u8, &[u8])>,
3107 ) -> Result<usize, FrameDecoderError> {
3108 let mut total_bytes_written = 0;
3109 while !input.is_empty() {
3110 match init_frame(self, &mut input) {
3111 Ok(_) => {}
3112 Err(FrameDecoderError::ReadFrameHeaderError(
3113 crate::decoding::errors::ReadFrameHeaderError::SkipFrame {
3114 magic_number,
3115 length,
3116 },
3117 )) => {
3118 let length = length as usize;
3119 // Visitor sees the payload slice BEFORE we advance
3120 // past it. Borrowed slice — no allocation. The
3121 // variant is the low nibble of the magic number
3122 // (RFC 8878 §3.1.2). `read_frame_header` only emits
3123 // SkipFrame for magic in 0x184D2A50..=0x184D2A5F, so
3124 // the subtraction fits in 0..=15.
3125 if input.len() < length {
3126 return Err(FrameDecoderError::FailedToSkipFrame);
3127 }
3128 let (payload, rest) = input.split_at(length);
3129 if let Some(visitor) = skippable_visitor.as_mut() {
3130 let variant = (magic_number - 0x184D2A50) as u8;
3131 visitor(variant, payload);
3132 }
3133 input = rest;
3134 continue;
3135 }
3136 Err(e) => return Err(e),
3137 };
3138 // Per-frame direct-path dispatch. Now safe to route the
3139 // public `decode_all` here because
3140 // `UserSliceBackend::exec_sequence_inline` returns
3141 // `Result<(), ExecuteSequencesError>` instead of
3142 // panicking on capacity overflow; the error propagates
3143 // up as `FrameDecoderError`. Eligibility (FCS > 0,
3144 // remaining `output` slice holds the declared content)
3145 // puts the frame on the fast path that bypasses the
3146 // FlatBuf/Ring -> `read()` drain copy. Ineligible frames
3147 // (no FCS, output too small) fall through to the legacy
3148 // `decode_blocks` + `read` drain loop below. Dictionary
3149 // frames are eligible (see the no-lsm path above).
3150 let (content_size, fcs_declared) = {
3151 let state_ref = self.state.as_ref().expect("init populated state");
3152 (
3153 state_ref.frame_header.frame_content_size(),
3154 state_ref.frame_header.fcs_declared(),
3155 )
3156 };
3157 // Only `cap >= frame_content_size` needed; the trailing
3158 // sequence(s) take the bounded copy in
3159 // `UserSliceBackend::exec_sequence_bounded`, so no
3160 // `WILDCOPY_OVERLENGTH` trailing slack is required (see the
3161 // no-lsm path above).
3162 // A frame that declares no size decodes straight into the
3163 // caller's slice too, the slice being its limit, as upstream
3164 // `ZSTD_decompressDCtx` decodes into `dst`. The drain path
3165 // reserved the frame's whole declared window for it, which for a
3166 // streamed producer's small frame is megabytes for kilobytes.
3167 let declared_size = fcs_declared.then_some(content_size);
3168 let direct_eligible = match declared_size {
3169 Some(declared) => declared > 0 && (output.len() as u64) >= declared,
3170 None => true,
3171 };
3172 if direct_eligible {
3173 let written = self.run_direct_decode(&mut input, output, declared_size)?;
3174 output = &mut output[written..];
3175 total_bytes_written += written;
3176 // Per-frame content-checksum verification (no-op unless the
3177 // mode is `Verify` and the frame carries a checksum).
3178 #[cfg(feature = "hash")]
3179 self.verify_content_checksum()?;
3180 continue;
3181 }
3182 // Non-direct fallback: pre-reserve the backing buffer to
3183 // `window_size` once so the per-block growth cycle is
3184 // skipped (see same comment on the no-lsm path above).
3185 if let Some(state) = self.state.as_mut() {
3186 // FCS-capped via `decoding_buffer_size` — the same cap
3187 // `decode_blocks` applies, so its per-iteration reserve in
3188 // the loop below cannot grow the buffer back to the raw
3189 // frame window.
3190 state.reserve_decoding_buffer();
3191 }
3192 let frame_start_total = total_bytes_written;
3193 loop {
3194 self.decode_blocks(&mut input, BlockDecodingStrategy::UptoBytes(1024 * 1024))?;
3195 let bytes_written = self
3196 .read(output)
3197 .map_err(FrameDecoderError::FailedToDrainDecodebuffer)?;
3198 output = &mut output[bytes_written..];
3199 total_bytes_written += bytes_written;
3200 if self.can_collect() != 0 {
3201 return Err(FrameDecoderError::TargetTooSmall);
3202 }
3203 if self.is_finished() {
3204 break;
3205 }
3206 }
3207 // Per-frame FCS validation on the legacy fallback path.
3208 // Use `fcs_declared()` (NOT `content_size > 0`) so an
3209 // empty frame with explicit FCS=0 on the wire still gets
3210 // validated.
3211 if fcs_declared {
3212 let produced = (total_bytes_written - frame_start_total) as u64;
3213 if produced != content_size {
3214 return Err(FrameDecoderError::FrameContentSizeMismatch {
3215 declared: content_size,
3216 produced,
3217 });
3218 }
3219 }
3220 // Per-frame content-checksum verification on the drain path: the
3221 // frame is fully decoded and drained here (is_finished + nothing
3222 // left to collect), so the running digest and stored value are
3223 // final. No-op unless the mode is `Verify`.
3224 #[cfg(feature = "hash")]
3225 self.verify_content_checksum()?;
3226 }
3227
3228 Ok(total_bytes_written)
3229 }
3230
3231 /// Decode multiple frames into the output slice using a serialized dictionary.
3232 ///
3233 /// # Warning
3234 ///
3235 /// Each decoded frame is initialized with the parsed dictionary, even when a
3236 /// frame header omits the optional dictionary ID. Callers must only use this
3237 /// API when they already know the input frames were encoded with that
3238 /// dictionary; otherwise decoded output can be silently corrupted.
3239 pub fn decode_all_with_dict_bytes(
3240 &mut self,
3241 input: &[u8],
3242 output: &mut [u8],
3243 raw_dictionary: &[u8],
3244 ) -> Result<usize, FrameDecoderError> {
3245 let dict = DictionaryHandle::decode_dict(raw_dictionary)?;
3246 self.decode_all_with_dict_handle(input, output, &dict)
3247 }
3248
3249 /// Decode multiple frames into the extra capacity of the output vector.
3250 ///
3251 /// `input` must contain an exact number of frames.
3252 ///
3253 /// `output` must have enough spare capacity to hold the decompressed
3254 /// data. This adds no extra slack: exact-fit output is now eligible
3255 /// for the direct decode path, so a `Vec::with_capacity(fcs)` is
3256 /// decoded straight into without a growth/reallocation. It will NOT
3257 /// grow the vector to fit the decompressed payload itself; the
3258 /// caller's pre-allocated capacity must already cover the data. If
3259 /// you don't know how large the output will be, use
3260 /// [`FrameDecoder::decode_blocks`] instead.
3261 ///
3262 /// This calls [`FrameDecoder::init`], and all bytes currently in the decoder will be lost.
3263 ///
3264 /// The length of the output vector is updated to include the
3265 /// decompressed data. The length is not changed if an error occurs.
3266 pub fn decode_all_to_vec(
3267 &mut self,
3268 input: &[u8],
3269 output: &mut Vec<u8>,
3270 ) -> Result<(), FrameDecoderError> {
3271 let len = output.len();
3272 let cap = output.capacity();
3273 output.resize(cap, 0);
3274 match self.decode_all(input, &mut output[len..]) {
3275 Ok(bytes_written) => {
3276 let new_len = core::cmp::min(len + bytes_written, cap); // Sanitizes `bytes_written`.
3277 output.resize(new_len, 0);
3278 Ok(())
3279 }
3280 Err(e) => {
3281 output.resize(len, 0);
3282 Err(e)
3283 }
3284 }
3285 }
3286
3287 /// Single-frame direct-decode path. Decodes one zstd frame into
3288 /// `output[..content_size]` via a stack-local
3289 /// `DecodeBuffer<UserSliceBackend>`, bypassing the per-block
3290 /// FlatBuf/Ring -> `read()` drain copy.
3291 ///
3292 /// # Preconditions (caller-enforced)
3293 ///
3294 /// - `self.init` (or `init_with_dict_handle`) was called for
3295 /// this frame so `self.state` is populated.
3296 /// - `declared_size` is the frame's declared content size, `> 0`, with
3297 /// `output.len() >= declared_size` (the eligibility gate), or `None`
3298 /// for a frame that declares none. Then `output` itself is the limit,
3299 /// as upstream `ZSTD_decompressDCtx` decodes into `dst`, and a frame
3300 /// that does not fit is `TargetTooSmall` rather than a size mismatch.
3301 /// No `WILDCOPY_OVERLENGTH` trailing slack is required: the trailing
3302 /// sequence(s) take the bounded (non-overshooting) copy in
3303 /// [`UserSliceBackend::exec_sequence_bounded`].
3304 ///
3305 /// Dictionary frames are supported: the scratch buffer's shared
3306 /// dict handle is forwarded to the stack-local `DecodeBuffer`, so
3307 /// offsets reaching past the frame's own output resolve through
3308 /// `repeat_from_dict` (the ext-dict slow path).
3309 ///
3310 /// On return, `input` points at the byte immediately after the
3311 /// frame's checksum (or after the last block, when the frame
3312 /// has `content_checksum_flag = 0`). `self.state.frame_finished`
3313 /// is set so [`Self::is_finished`] reports `true`.
3314 fn run_direct_decode(
3315 &mut self,
3316 input: &mut &[u8],
3317 output: &mut [u8],
3318 declared_size: Option<u64>,
3319 ) -> Result<usize, FrameDecoderError> {
3320 #[cfg(test)]
3321 {
3322 self.direct_frames += 1;
3323 }
3324 use super::block_decoder;
3325 use super::decode_buffer::DecodeBuffer;
3326 use super::scratch::DirectScratch;
3327 use super::user_slice_buf::UserSliceBackend;
3328 use crate::io::Read;
3329 use FrameDecoderError as err;
3330
3331 // The most the frame may write: its declared size, or the caller's
3332 // slice for a frame that declares none.
3333 let limit = declared_size.unwrap_or(output.len() as u64);
3334 // Output past `limit`: the frame lied about its size, or it does not
3335 // fit the caller's slice.
3336 let overflow = |produced: u64| match declared_size {
3337 Some(declared) => err::FrameContentSizeMismatch { declared, produced },
3338 None => err::TargetTooSmall,
3339 };
3340
3341 let kernel = self.kernel;
3342 let state = self
3343 .state
3344 .as_mut()
3345 .expect("caller ensures init populated state");
3346
3347 // Fast path: a frame that is a single RAW block spanning the whole
3348 // declared content. Upstream zstd handles this as one `ZSTD_copyRawBlock`
3349 // (a `memmove`) inside `ZSTD_decompressFrame`; do the same here — a direct
3350 // `copy_from_slice` into the caller's output slice — skipping the
3351 // `DirectScratch` / `DecodeBuffer` / `UserSliceBackend` wrapper
3352 // construction and the general per-block loop. Incompressible payloads
3353 // (random / already-compressed data) emit exactly this shape, so the win
3354 // lands on small high-entropy frames where the per-frame machinery, not
3355 // the 1-block copy, dominates.
3356 {
3357 let mut probe = *input;
3358 let mut header_dec = block_decoder::with_kernel(kernel);
3359 if let Ok((bh, hsize)) = header_dec.read_block_header(&mut probe) {
3360 let n = bh.decompressed_size as usize;
3361 // A frame that declares no size takes the shortcut too: the
3362 // slice is its limit, and holding the block means holding the
3363 // frame. Without this the probe parsed the header that the
3364 // general loop below parses again, on the very path (a small
3365 // frame from a streamed producer) this decode is for. The
3366 // block maximum is checked here as the general path checks it:
3367 // a block past it is malformed, and the shortcut must not be
3368 // the way around that.
3369 let window = state.frame_header.window_size().unwrap_or(0) as usize;
3370 if bh.last_block
3371 && matches!(bh.block_type, crate::blocks::block::BlockType::Raw)
3372 && declared_size.is_none_or(|declared| declared == n as u64)
3373 && n <= block_decoder::block_maximum(window)
3374 && probe.len() >= n
3375 && output.len() >= n
3376 {
3377 output[..n].copy_from_slice(&probe[..n]);
3378 *input = &probe[n..];
3379 state.bytes_read_counter += u64::from(hsize) + n as u64;
3380 state.block_counter += 1;
3381 // Consume the trailing 4-byte content checksum UNCONDITIONALLY
3382 // when the frame declares one — exactly like the general
3383 // direct loop and `decode_blocks`. Only the hash SEEDING is
3384 // `hash`-gated; the byte consumption / counter / `check_sum`
3385 // must not be, or a no-`hash` build leaves the 4 bytes in
3386 // `*input` (misparsed as the next frame) and never sets
3387 // `check_sum` (so `is_finished` stays false).
3388 if state.frame_header.descriptor.content_checksum_flag() {
3389 let mut chksum = [0u8; 4];
3390 Read::read_exact(input, &mut chksum).map_err(err::FailedToReadChecksum)?;
3391 state.bytes_read_counter += 4;
3392 state.check_sum = Some(u32::from_le_bytes(chksum));
3393 // Mirror the general path: seed the scratch hash so
3394 // `verify_content_checksum` / `get_calculated_checksum`
3395 // read the digest. Skipped under `ContentChecksum::None`.
3396 #[cfg(feature = "hash")]
3397 if self.content_checksum != ContentChecksum::None {
3398 use core::hash::Hasher;
3399 let mut h = twox_hash::XxHash64::with_seed(0);
3400 h.write(&output[..n]);
3401 match &mut state.decoder_scratch {
3402 DecoderScratchKind::Flat(s) => s.buffer.set_hash(h),
3403 DecoderScratchKind::Ring(s) => s.buffer.set_hash(h),
3404 }
3405 }
3406 }
3407 #[cfg(all(feature = "lsm", feature = "hash"))]
3408 if self.per_block_checksums_enabled {
3409 use core::hash::Hasher;
3410 let mut h = twox_hash::XxHash64::with_seed(0);
3411 h.write(&output[..n]);
3412 self.computed_block_checksums.push(h.finish() as u32);
3413 }
3414 state.frame_finished = true;
3415 return Ok(n);
3416 }
3417 }
3418 }
3419
3420 // Borrow persistent fields out of whichever scratch variant
3421 // `init` produced (Flat for single_segment, Ring for
3422 // multi-segment) — both expose the same HUF/FSE/Vec
3423 // fields; only `buffer` differs and we don't use that here.
3424 // Macro-style binding avoids the closure / generic
3425 // gymnastics of returning multiple `&mut` from a match arm.
3426 // Resolve the dictionary borrow for this frame BEFORE taking the
3427 // `&mut` field borrows below — `active_dict` is a disjoint field, so
3428 // the shared borrow coexists with the mutable scratch borrows. It is
3429 // threaded as a call-scoped argument into every `Dict`-sourced read
3430 // (the direct path's `repeat_from_dict` ext-dict slow path), mirroring
3431 // C's per-frame pointer hand-off with zero refcount churn.
3432 // Only expose the held dictionary while THIS frame is dict-backed
3433 // (`using_dict` is set per dict-apply, cleared on reset). A reused
3434 // decoder keeps `active_dict` across a no-dict frame for the
3435 // `ptr::eq` reuse-skip, so it must be gated here or a stray
3436 // out-of-window offset on a dictless frame would resolve against the
3437 // stale dictionary content instead of erroring.
3438 let dict_ref = if state.using_dict.is_some() {
3439 state.active_dict.as_ref().map(|h| h.as_dict())
3440 } else {
3441 None
3442 };
3443 let (huf, fse, offset_hist, literals_buffer, block_content_buffer, window_size) =
3444 match &mut state.decoder_scratch {
3445 DecoderScratchKind::Flat(s) => (
3446 &mut s.huf,
3447 &mut s.fse,
3448 &mut s.offset_hist,
3449 &mut s.literals_buffer,
3450 &mut s.block_content_buffer,
3451 s.buffer.window_size,
3452 ),
3453 DecoderScratchKind::Ring(s) => (
3454 &mut s.huf,
3455 &mut s.fse,
3456 &mut s.offset_hist,
3457 &mut s.literals_buffer,
3458 &mut s.block_content_buffer,
3459 s.buffer.window_size,
3460 ),
3461 };
3462 let backend = UserSliceBackend::from_slice(output);
3463 let buffer = DecodeBuffer::from_backend(backend, window_size);
3464 let mut direct = DirectScratch {
3465 huf,
3466 fse,
3467 offset_hist,
3468 literals_buffer,
3469 block_content_buffer,
3470 buffer,
3471 };
3472
3473 // Block loop. Mirrors `decode_blocks` (without the
3474 // strategy-bounded early exit — we always decode the whole
3475 // frame in one shot for the direct path). Keeps
3476 // `state.bytes_read_counter` / `state.block_counter` in
3477 // sync with `decode_blocks` so post-call accessors
3478 // (`bytes_read_from_source`, `blocks_decoded`) return
3479 // accurate values.
3480 let mut block_dec = block_decoder::with_kernel(kernel);
3481 // Track total output bytes against the declared
3482 // `frame_content_size` via the buffer's actual write
3483 // counter — `BlockHeader.decompressed_size` is 0 for
3484 // Compressed blocks (the header parser can't know the
3485 // expanded size before decoding the body), so per-header
3486 // tracking would always count 0 for those blocks and
3487 // miscount frames that aren't pure Raw/RLE.
3488 let mut produced: u64 = 0;
3489 // Per-block output ranges captured during the direct-path
3490 // loop. After the loop we re-borrow `output` (post-drop of
3491 // `direct`) and XXH64 each range into
3492 // `self.computed_block_checksums`, so the digests vector
3493 // stays consistent with the legacy `decode_blocks` path
3494 // regardless of which dispatch the frame took.
3495 // `Vec::new()` does not allocate, so this stays free when
3496 // `per_block_checksums_enabled` is false: the `push` and the
3497 // post-loop hashing loop are both gated by the same flag.
3498 #[cfg(all(feature = "lsm", feature = "hash"))]
3499 let mut block_ranges: alloc::vec::Vec<(usize, usize)> = alloc::vec::Vec::new();
3500 // Frame-level XXH64, accumulated PER BLOCK right after each block
3501 // decodes — the bytes are still cache-resident then. The previous
3502 // shape hashed the whole output once after the loop, which re-read
3503 // the entire frame cold: a full extra memory pass that the
3504 // reference implementation does not make (it hashes incrementally
3505 // per block). Invisible on outputs that fit L3, ~1.14x wall on a
3506 // 100 MiB all-raw decode and the dominant CI gap on
3507 // bandwidth-limited hosts.
3508 #[cfg(feature = "hash")]
3509 let mut running_hash: Option<twox_hash::XxHash64> =
3510 if state.frame_header.descriptor.content_checksum_flag()
3511 && self.content_checksum != ContentChecksum::None
3512 {
3513 Some(twox_hash::XxHash64::with_seed(0))
3514 } else {
3515 None
3516 };
3517 loop {
3518 #[cfg(all(feature = "lsm", feature = "hash"))]
3519 let produced_before: Option<usize> = if self.per_block_checksums_enabled {
3520 Some(produced as usize)
3521 } else {
3522 None
3523 };
3524 // Failing-block coordinates captured before the header read (see
3525 // the `decode_blocks` loop for the rationale).
3526 let block_index = state.block_counter as u32;
3527 let block_frame_offset = state.bytes_read_counter as u32;
3528 let (block_header, hsize) =
3529 block_dec.read_block_header(&mut *input).map_err(|source| {
3530 block_header_decode_error(source, block_index, block_frame_offset)
3531 })?;
3532 state.bytes_read_counter += u64::from(hsize);
3533 // Pre-flight FCS check ONLY for Raw / RLE blocks where
3534 // `decompressed_size` is the actual block output size.
3535 // For Compressed blocks the header field is 0; the
3536 // post-decode check below catches overflow via the
3537 // backend's actual write counter delta.
3538 let block_upper = u64::from(block_header.decompressed_size);
3539 if block_upper > 0 && produced + block_upper > limit {
3540 // Raw/RLE block headers claim more output than the FCS
3541 // allows (a corrupt frame) or the caller's slice holds.
3542 return Err(overflow(produced + block_upper));
3543 }
3544 // Slice-source fast path: consume the block body
3545 // straight from `input` without copying into the
3546 // persistent `block_content_buffer`.
3547 let body_consumed = match block_dec.decode_block_content_from_slice(
3548 &block_header,
3549 &mut direct,
3550 dict_ref,
3551 &mut *input,
3552 ) {
3553 Ok(n) => n,
3554 // Defense-in-depth: RLE / Raw block whose declared
3555 // `decompressed_size` slipped past the per-block
3556 // pre-flight above and tripped the backend's
3557 // fallible write surface.
3558 Err(crate::decoding::errors::DecodeBlockContentError::BackendOverflow {
3559 ..
3560 }) => {
3561 // Use saturating_add on the
3562 // `produced + decompressed_size` sum. Each block
3563 // is bounded by 128 KiB (MAX_BLOCK_SIZE), but
3564 // accumulated `produced` can grow toward
3565 // u64::MAX across adversarial frames. Saturating
3566 // avoids a panic on the error path itself.
3567 return Err(overflow(
3568 produced.saturating_add(u64::from(block_header.decompressed_size)),
3569 ));
3570 }
3571 // Compressed-block in-block overshoot: the sequence
3572 // executor (upstream zstd-inline path) or the match-repeat
3573 // fallback tripped the fixed-capacity backend's per-write
3574 // check. Unlike Raw/RLE, a Compressed block carries no
3575 // header-declared output size, so `produced` is computed
3576 // from the partial fill: `tail` bytes were written before
3577 // the failing op, and `requested` is what overflowed —
3578 // their sum is a strict lower bound on the frame's true
3579 // expanded size and is always > `limit`. With a declared
3580 // size the slice holds at least that much, so any overflow
3581 // means the frame exceeded its FCS, never a caller-undersized
3582 // buffer, and folds into the same `FrameContentSizeMismatch`
3583 // contract as Raw/RLE; without one the slice is the limit.
3584 // An overflow that stays within `limit` was refused by the
3585 // per-block output ceiling instead: a malformed block, which
3586 // takes the generic arm below.
3587 Err(crate::decoding::errors::DecodeBlockContentError::DecompressBlockError(
3588 crate::decoding::errors::DecompressBlockError::ExecuteSequencesError(ref e),
3589 )) if e.output_overflow_requested().is_some_and(|requested| {
3590 (direct.buffer.buffer_ref().tail() as u64).saturating_add(requested as u64)
3591 > limit
3592 }) =>
3593 {
3594 let requested = e
3595 .output_overflow_requested()
3596 .expect("guard guarantees Some") as u64;
3597 let tail = direct.buffer.buffer_ref().tail() as u64;
3598 return Err(overflow(tail.saturating_add(requested)));
3599 }
3600 // A no-sequence block's literals did not fit the slice. Every
3601 // direct-path entry holds `output.len() >= limit` (a declared
3602 // size is checked against the slice before the path is chosen,
3603 // and an undeclared frame's limit IS the slice), so a write
3604 // past the slice is a write past `limit`: the frame outgrew its
3605 // declared size, or the caller's target is short.
3606 Err(crate::decoding::errors::DecodeBlockContentError::DecompressBlockError(
3607 crate::decoding::errors::DecompressBlockError::LiteralsOutputOverflow {
3608 tail,
3609 requested,
3610 capacity,
3611 },
3612 )) => {
3613 debug_assert!(
3614 capacity as u64 >= limit,
3615 "direct path entered with a short slice"
3616 );
3617 return Err(overflow((tail as u64).saturating_add(requested as u64)));
3618 }
3619 Err(e) => {
3620 return Err(block_body_decode_error(
3621 e,
3622 block_index,
3623 block_frame_offset,
3624 &block_header,
3625 hsize,
3626 ));
3627 }
3628 };
3629 // Hash this block's freshly-written bytes while they are hot
3630 // (see `running_hash` above). `tail()` is the physical write
3631 // cursor: `drop_to_window_size` below only advances the head,
3632 // so `[prev_tail, tail)` is exactly this block's output.
3633 #[cfg(feature = "hash")]
3634 if let Some(hasher) = running_hash.as_mut() {
3635 use core::hash::Hasher;
3636 hasher.write(direct.buffer.buffer_ref().written_since(produced as usize));
3637 }
3638 produced = direct.buffer.buffer_ref().tail() as u64;
3639 // Post-decode FCS overflow check.
3640 if produced > limit {
3641 return Err(overflow(produced));
3642 }
3643 state.bytes_read_counter += body_consumed;
3644 state.block_counter += 1;
3645 #[cfg(all(feature = "lsm", feature = "hash"))]
3646 if let Some(produced_before) = produced_before {
3647 block_ranges.push((produced_before, produced as usize));
3648 }
3649 // Cap the visible buffer at window_size between blocks
3650 // so the next block's match-offset validation matches
3651 // the spec's `offset <= window_size` rule.
3652 direct.buffer.drop_to_window_size();
3653 if block_header.last_block {
3654 if state.frame_header.descriptor.content_checksum_flag() {
3655 let mut chksum = [0u8; 4];
3656 input
3657 .read_exact(&mut chksum)
3658 .map_err(err::FailedToReadChecksum)?;
3659 state.bytes_read_counter += 4;
3660 state.check_sum = Some(u32::from_le_bytes(chksum));
3661 }
3662 break;
3663 }
3664 }
3665 // Final sanity: blocks summed to exactly the declared size.
3666 if let Some(declared) = declared_size
3667 && produced != declared
3668 {
3669 return Err(err::FrameContentSizeMismatch { declared, produced });
3670 }
3671
3672 let written = produced as usize;
3673 state.frame_finished = true;
3674 // `direct`'s last use is in the decode loop above; NLL therefore
3675 // releases its `&mut output` borrow before here, freeing `output` for
3676 // the hash re-borrow below. No explicit `drop(direct)` is needed:
3677 // `DirectScratch` now holds only borrowed dict POINTERS (not an owned
3678 // `Arc`), so it is not a `Drop` type whose glue would hold the borrow
3679 // to end-of-scope.
3680 // Per-block XXH64 (low 32 bits) over the captured ranges.
3681 // Mirrors `decode_blocks`' per-block hashing so the digests
3682 // vector stays identical regardless of which dispatch path
3683 // the frame took. Ranges were recorded inside the loop while
3684 // `direct` held a mutable borrow on `output`; now that the
3685 // borrow is dropped we can read the slices directly.
3686 #[cfg(all(feature = "lsm", feature = "hash"))]
3687 if self.per_block_checksums_enabled {
3688 use core::hash::Hasher;
3689 for (start, end) in &block_ranges {
3690 let mut h = twox_hash::XxHash64::with_seed(0);
3691 h.write(&output[*start..*end]);
3692 self.computed_block_checksums.push(h.finish() as u32);
3693 }
3694 }
3695 #[cfg(feature = "hash")]
3696 if let Some(hasher) = running_hash {
3697 // Propagate the per-block-accumulated hasher state (see the
3698 // `running_hash` rationale above the loop) so the frame-tail
3699 // XXH64 check and `get_calculated_checksum()` read the digest.
3700 // `running_hash` is `None` for flag-off frames or
3701 // `ContentChecksum::None` — nothing to verify there, and
3702 // `get_calculated_checksum()` returns `None`, matching the skip.
3703 match &mut state.decoder_scratch {
3704 DecoderScratchKind::Flat(s) => s.buffer.set_hash(hasher),
3705 DecoderScratchKind::Ring(s) => s.buffer.set_hash(hasher),
3706 }
3707 }
3708 Ok(written)
3709 }
3710}
3711
3712/// Read bytes from the decode_buffer that are no longer needed. While the frame is not yet finished
3713/// this will retain window_size bytes, else it will drain it completely
3714impl Read for FrameDecoder {
3715 fn read(&mut self, target: &mut [u8]) -> Result<usize, Error> {
3716 let state = match &mut self.state {
3717 None => return Ok(0),
3718 Some(s) => s,
3719 };
3720 if state.frame_finished {
3721 state.decoder_scratch.buffer_read_all(target)
3722 } else {
3723 state.decoder_scratch.buffer_read(target)
3724 }
3725 }
3726}
3727
3728#[cfg(test)]
3729mod tests;