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