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