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