coordinode-lsm-tree 5.5.0

// SPDX-License-Identifier: Apache-2.0
// Copyright (c) 2025-present, fjall-rs
// Copyright (c) 2026-present, Structured World Foundation

pub(crate) mod binary_index;
// Crate-internal: Decoder, Decodable, ParsedItem are not part of the public API.
// They are re-exported as pub(crate) below; narrowing the module prevents
// external code from reaching these traits via lsm_tree::table::block::decoder::*.
pub(crate) mod decoder;
mod encoder;
pub mod hash_index;
pub(crate) mod header;
mod identity;
pub(crate) mod kv_checksum;
mod offset;
mod trailer;
mod transform;
mod r#type;

pub(crate) use decoder::{Decodable, Decoder, DecoderMeta, ParsedItem};
pub(crate) use encoder::{Encodable, Encoder};
pub use header::Header;
pub use identity::BlockIdentity;
pub use offset::BlockOffset;
pub(crate) use trailer::{TRAILER_START_MARKER, Trailer};
pub use transform::{BlockTransform, CompressionContext, EccParams};
pub use r#type::BlockType;

#[cfg(zstd_any)]
use crate::compression::CompressionProvider as _;

use crate::{
    Checksum, CompressionType, Slice,
    coding::{Decode, Encode},
    fs::FsFile,
    table::BlockHandle,
};
use alloc::borrow::Cow;
// Vec lives in the std prelude on std builds; pull it from alloc on no-std so
// this module's heap buffers resolve under `--no-default-features --features alloc`.
#[cfg(not(feature = "std"))]
use alloc::vec::Vec;

/// Safety cap on block payload size (256 MiB).
///
/// Enforced on both write and read paths to prevent producing or accepting
/// blocks that are unreasonably large. Intentionally stricter than the
/// on-disk format limit (`u32::MAX`) to guard against decompression bombs
/// and OOM from crafted/malicious SST files.
///
/// NOTE: Intentionally duplicated in `vlog::blob_file` (writer as `usize`,
/// reader as `usize`) rather than shared, because blocks and blobs are
/// independent storage formats that may diverge in the future.
const MAX_DECOMPRESSION_SIZE: u32 = 256 * 1024 * 1024;

/// Exact parity trailer size for a given data length under the supplied
/// shard scheme `params` (`data_shards`, `parity_shards`).
///
/// `parity_len(N) = shard_bytes(N) * parity_shards` where
/// `shard_bytes(N) = ceil(N / data_shards) rounded up to the nearest even`.
/// Mirrors `crate::ecc::parity_len` byte-for-byte but is available
/// without the `page_ecc` cargo feature — the parity trailer length is
/// NOT stored in the block header; the read path derives it from
/// `data_length` and the per-SST scheme whenever a block's `ECC_PARITY`
/// flag is set, so writer and reader always agree on the trailer size
/// without a per-block length field to corrupt or forge.
///
/// Returns zero for `data_length == 0` to match
/// `crate::ecc::encode_parity`'s empty-payload short-circuit.
#[inline]
pub(crate) fn expected_parity_len(data_length: u32, params: EccParams) -> u32 {
    // SEC-DED: one parity byte per 8-byte word (matches
    // `crate::secded::block_parity_len`); no shard arithmetic.
    let (data_shards, parity_shards) = match params {
        EccParams::Secded => return data_length.div_ceil(8),
        EccParams::Shard {
            data_shards,
            parity_shards,
        } => (u32::from(data_shards), u32::from(parity_shards)),
    };
    if data_length == 0 || data_shards == 0 || parity_shards == 0 {
        return 0;
    }
    // ceil(N / data_shards). The division never increases the value and the
    // remainder bump only fires when there IS a remainder (so the quotient is
    // already below the dividend), keeping this within u32 — plain arithmetic.
    let ceil = (data_length / data_shards) + u32::from(!data_length.is_multiple_of(data_shards));
    // Round up to even (the `reed-solomon-simd` engine requires shard sizes that
    // are a multiple of two; XOR shares the layout). With `data_shards == 1` and
    // a `u32::MAX` data length, `ceil` is an odd `u32::MAX`, so the +1 must
    // saturate; a corrupt header reaching here is rejected downstream.
    let shard_bytes = ceil.saturating_add(u32::from(!ceil.is_multiple_of(2)));
    // `parity_shards` is a u8 (≤ 255), so for a large block with many parity
    // shards the product CAN exceed u32 — saturate. An over-large parity length
    // is rejected against the actual block downstream, so the clamp is safe.
    shard_bytes.saturating_mul(parity_shards)
}

/// Whether the on-disk block carries a Reed-Solomon parity trailer.
///
/// Source of truth depends on the block type:
/// - Blocks that carry the `block_flags` byte (`Meta` / `Manifest` /
///   `ManifestFooter` — see [`Header::has_block_flags`]) self-describe parity
///   via the `ECC_PARITY` bit.
/// - SST blocks (`Data` / `Index` / `Filter` / `RangeTombstone`) omit the
///   byte, so
///   parity presence comes from the per-SST descriptor, threaded in via the
///   caller-supplied `transform` (`transform.page_ecc()`).
fn block_has_parity(header: &Header, transform: &BlockTransform<'_>) -> bool {
    if Header::has_block_flags(header.block_type) {
        header.block_flags & header::block_flags::ECC_PARITY != 0
    } else {
        transform.page_ecc()
    }
}

/// The ECC shard scheme to size + recover a block's parity trailer with.
///
/// Self-describing blocks (`Meta` / `Manifest` / `ManifestFooter`) are read
/// at table / manifest open BEFORE any per-SST descriptor is known, so they
/// always use the fixed [`EccParams::RS_4_2`] layout (matching the writer).
/// SST blocks (`Data` / `Index` / `Filter` / `RangeTombstone`) are
/// descriptor-driven: their scheme rides on the caller-supplied `transform`
/// (sourced from the SST's `TableMeta`).
///
/// Not feature-gated: callable from read sizing on all builds (without
/// `page_ecc` it is only reached from dead `block_has_parity == false`
/// branches, but must still compile).
fn block_ecc_params(header: &Header, transform: &BlockTransform<'_>) -> EccParams {
    if Header::has_block_flags(header.block_type) {
        EccParams::RS_4_2
    } else {
        transform.ecc_params().unwrap_or(EccParams::RS_4_2)
    }
}

/// On-disk `CompressionType` tag byte (spec §5.1 registry: 0 none, 1 lz4,
/// 3 zstd, 4 zstd+dict). Used as the AAD-bound `compression_type` field so a
/// codec relabel (e.g. forging a different window/dict context) fails AEAD
/// verification. Mirrors the leading byte of [`CompressionType`]'s `Encode`.
#[cfg(zstd_any)]
fn compression_tag_byte(compression: CompressionType) -> u8 {
    match compression {
        CompressionType::None => 0,
        #[cfg(feature = "lz4")]
        CompressionType::Lz4 => 1,
        CompressionType::Zstd(_) => 3,
        CompressionType::ZstdDict { .. } => 4,
    }
}

/// Seal a (post-compression) block payload for the on-disk encryption envelope.
///
/// Under zstd builds this is the AAD-bound `MetadataFrame ‖ BodyFrame` envelope
/// ([`EncryptionProvider::encrypt_block_aad`]) binding the block identity +
/// transform context. Without zstd the wire-format frame is unavailable, so it
/// falls back to the opaque `[nonce ‖ ciphertext ‖ tag]` form. `owned` is the
/// compressor's output (reused in-place on the opaque path); `borrow` is the
/// raw `data` for the uncompressed case.
#[cfg_attr(
    zstd_any,
    expect(
        clippy::needless_pass_by_value,
        reason = "owned is consumed by encrypt_vec on the non-zstd path; the AAD \
                  path only borrows it, so by-value is needed for the other cfg"
    )
)]
fn encrypt_block_payload(
    enc: &dyn crate::encryption::EncryptionProvider,
    owned: Option<Vec<u8>>,
    borrow: &[u8],
    identity: &BlockIdentity,
    compression: CompressionType,
    block_flags: u8,
) -> crate::Result<Vec<u8>> {
    #[cfg(zstd_any)]
    {
        let plaintext = owned.as_deref().unwrap_or(borrow);
        // The `ECC_PARITY` flag describes the Reed-Solomon parity trailer,
        // which is OUTER framing: parity is computed over the (encrypted)
        // payload AFTER this seal and stripped BEFORE decrypt, so the bit is
        // not part of the encrypted content and must not enter the AAD. It is
        // also unknown here (set only once the trailer is emitted, later in
        // `prepare_with_flags`). Masking it keeps seal == verify regardless of
        // pipeline ordering; its integrity is self-enforced (a flipped bit
        // mis-strips the trailer, so the AEAD then runs over wrong bytes and
        // fails). The plaintext-affecting transform bits (COMPRESSED /
        // ENCRYPTED / KV_CHECKSUM_FOOTER) stay bound.
        let aad_block_flags = block_flags & !crate::table::block::header::block_flags::ECC_PARITY;
        enc.encrypt_block_aad(
            plaintext,
            identity,
            compression_tag_byte(compression),
            aad_block_flags,
        )
    }
    #[cfg(not(zstd_any))]
    {
        let _ = (identity, compression, block_flags);
        match owned {
            Some(buf) => enc.encrypt_vec(buf),
            None => enc.encrypt(borrow),
        }
    }
}

/// Inverse of [`encrypt_block_payload`]: recover the plaintext from the on-disk
/// envelope. Under zstd builds it verifies the AAD binding via
/// [`EncryptionProvider::decrypt_block_aad`] (the reader supplies only
/// `identity`; the transform fields are read back from the frame); without zstd
/// it falls back to the opaque in-place `decrypt_vec`.
fn decrypt_block_payload(
    enc: &dyn crate::encryption::EncryptionProvider,
    raw: &[u8],
    identity: &BlockIdentity,
) -> crate::Result<Vec<u8>> {
    #[cfg(zstd_any)]
    {
        enc.decrypt_block_aad(raw, identity)
    }
    #[cfg(not(zstd_any))]
    {
        let _ = identity;
        // `decrypt_vec` consumes an owned buffer; the read path now hands us a
        // borrowed Slice, so copy once (this branch is the no-zstd build only).
        enc.decrypt_vec(raw.to_vec())
    }
}

/// Classifies the on-disk trailer (bytes after the `data_length` payload) into
/// an [`EccStatus`], or `Err` on a framing violation.
///
/// The decision keys off whether the block carries a RECOGNIZED ECC layout
/// (`has_recognized_ecc`, from [`block_has_parity`]), NOT off `ecc_length`:
/// a recognized layout on an empty payload also has `ecc_length == 0`, and its
/// trailer must still be exactly zero. So:
/// - recognized layout → require `trailer == ecc_length` exactly (extra bytes
///   are corruption), even when `ecc_length == 0`;
/// - no recognized layout, `trailer == 0` → ECC off, clean;
/// - no recognized layout, `trailer > 0` → an ECC trailer this build can't
///   interpret: the payload is framed by `data_length` and verified by its
///   checksum, so the read succeeds with [`EccStatus::Unrecognized`] (a WARN),
///   but recovery is unavailable;
/// - `actual_payload_plus_ecc < data_length` (payload doesn't fit) → corruption.
fn classify_block_trailer(
    has_recognized_ecc: bool,
    actual_payload_plus_ecc: usize,
    data_length: usize,
    ecc_length: u32,
    handle: &BlockHandle,
) -> crate::Result<EccStatus> {
    let trailer = actual_payload_plus_ecc
        .checked_sub(data_length)
        .ok_or(crate::Error::InvalidHeader("Block"))?;
    if has_recognized_ecc {
        if trailer != ecc_length as usize {
            return Err(crate::Error::InvalidHeader("Block"));
        }
        Ok(EccStatus::Ok)
    } else if trailer == 0 {
        Ok(EccStatus::Ok)
    } else {
        log::warn!(
            "block {handle:?} carries an unrecognized ECC trailer ({trailer} B); \
             payload verified by checksum but recovery is unavailable — recompact \
             to re-stamp with a supported scheme",
        );
        Ok(EccStatus::Unrecognized)
    }
}

/// A block whose transform pipeline (compress → encrypt → checksum → ecc)
/// has run, but whose framed bytes have not yet been written to the file.
///
/// Splitting "prepare" from "write" is what lets compaction run the
/// CPU-bound transform stack on worker threads while a single thread keeps
/// the file writes (and the index registration that depends on byte offsets)
/// strictly ordered. The serial path stays zero-copy: `payload` borrows the
/// caller's `data` on the uncompressed/unencrypted path and owns it only when
/// a transform produced a fresh buffer. A worker thread takes ownership via
/// [`PreparedBlock::into_owned`] so the prepared block can outlive `data`.
pub(crate) struct PreparedBlock<'a> {
    header: Header,
    payload: Cow<'a, [u8]>,
    /// Reed-Solomon parity trailer, present only when page-ECC is active and
    /// the payload was non-empty. Always `None` without the `page_ecc` feature.
    parity: Option<Vec<u8>>,
    /// Inner zstd-block layout of the compressed payload: cumulative
    /// decompressed END offsets, one per inner block (the last equals the
    /// uncompressed length). Empty unless this is a `Data` block compressed
    /// into >= 2 inner zstd blocks; lets the reader partial-decode a key-range
    /// subset instead of the whole block. See
    /// [`CompressionProvider::compress_with_layout`](crate::compression::CompressionProvider::compress_with_layout).
    pub(crate) layout: Vec<u32>,
}

impl PreparedBlock<'_> {
    /// Takes ownership of the payload so the prepared block no longer borrows
    /// the source `data`, yielding a `'static` value safe to move to a worker
    /// thread. A no-op allocation when the payload is already owned (a
    /// transform ran); copies once on the borrowed (uncompressed) path.
    #[cfg(feature = "std")] // no-std: parallel compaction unavailable (no threads)
    pub(crate) fn into_owned(self) -> PreparedBlock<'static> {
        PreparedBlock {
            header: self.header,
            payload: Cow::Owned(self.payload.into_owned()),
            parity: self.parity,
            layout: self.layout,
        }
    }

    /// Writes the framed block (header + payload + optional parity trailer)
    /// to `writer` and returns the header. This is the single point where
    /// block bytes hit the file, so it must run in on-disk order.
    pub(crate) fn write_to<W: crate::io::Write>(self, mut writer: &mut W) -> crate::Result<Header> {
        self.header.encode_into(&mut writer)?;
        writer.write_all(&self.payload)?;
        if let Some(parity) = &self.parity {
            writer.write_all(parity)?;
        }

        log::trace!(
            "Writing block with size {}B (on-disk: {}B, ecc: {}B) (excluding header of {}B)",
            self.header.uncompressed_length,
            self.header.data_length,
            self.parity.as_ref().map_or(0, Vec::len),
            Header::header_len(self.header.block_type),
        );

        Ok(self.header)
    }
}

/// A block on disk
///
/// Consists of a fixed-size header and some bytes (the data/payload).
/// Outcome of the ECC check performed while reading a block, distinct from
/// success / failure of the read itself.
///
/// A read returns `Err` only on real payload corruption (checksum mismatch
/// with no available recovery). When the read *succeeds*, this reports
/// whether the block's ECC was usable:
///
/// - [`Self::Ok`] — the read returned correct bytes with no ECC intervention:
///   ECC was absent, or a recognized scheme verified the payload as-is (no
///   repair needed).
/// - [`Self::Corrected`] — ECC repaired the on-disk payload (the caller saw
///   correct bytes, but the on-disk copy still holds a latent fault). Treat as
///   a signal to confirm persistence and potentially schedule an auto-heal
///   recompaction. Which mechanism did the repair (SEC-DED vs RS shard) is
///   surfaced separately to internal read paths via [`EccRecoveryKind`] for
///   metrics attribution.
/// - [`Self::Unrecognized`] — the block carries an ECC trailer this build
///   cannot interpret (a non-canonical scheme, page granularity, unknown kind,
///   …). The payload was returned (its checksum passed), but ECC recovery is
///   unavailable for this block; recompaction re-stamps it with a supported
///   scheme. A "typing" warning, not a read failure.
#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash, Default)]
pub enum EccStatus {
    /// ECC absent or a recognized scheme verified normally.
    #[default]
    Ok,
    /// ECC trailer present but its scheme is unrecognized / unusable; the
    /// payload still verified by its checksum. Recommend recompaction.
    Unrecognized,
    /// The on-disk payload failed its checksum and was REPAIRED by ECC
    /// (SEC-DED single-bit heal or Reed-Solomon shard recovery): the bytes
    /// returned to the caller are correct (they reproduce the stored
    /// checksum), but the on-disk copy still holds the latent fault. A signal
    /// for auto-heal — the caller may re-read to confirm the corruption is
    /// persistent (not a transient read-path fault) and, if so, schedule a
    /// recompaction so the corrected bytes are persisted to a fresh SST.
    Corrected,
}

/// Which ECC mechanism recovered a block whose checksum failed on read.
///
/// Returned alongside [`EccStatus`] by the internal recovery-aware read paths
/// (kept out of the public `EccStatus` to keep that enum stable) so an operator
/// metric can attribute each on-read recovery to the right heal path: the cheap
/// single-bit fast path versus full shard reconstruction.
#[derive(Copy, Clone, Debug, Eq, PartialEq, Hash)]
pub enum EccRecoveryKind {
    /// A single-bit flip healed by the SEC-DED fast path (one XOR-and-recheck
    /// per word, no shard arithmetic).
    Secded,
    /// Recovered from Reed-Solomon shard parity (the general multi-byte path).
    Shard,
}

#[derive(Clone)]
pub struct Block {
    pub header: Header,
    pub data: Slice,
}

impl Block {
    /// Returns the uncompressed block size in bytes.
    #[must_use]
    pub fn size(&self) -> usize {
        self.data.len()
    }

    /// Reads `data_length` payload bytes, then `ecc_length` parity
    /// bytes (when non-zero), verifies the payload checksum against
    /// `expected`, and on mismatch attempts Reed-Solomon recovery
    /// from the parity trailer. Returns the validated payload bytes
    /// (recovered if needed).
    ///
    /// Always consumes exactly `data_length + ecc_length` bytes from
    /// the reader, so callers don't have to track the trailer
    /// independently. When ECC recovery succeeds, the original
    /// checksum-mismatch is logged at WARN level — the block is
    /// returned to the caller as if no corruption ever happened.
    fn read_payload_and_verify<R: crate::io::Read>(
        reader: &mut R,
        data_length: u32,
        ecc_length: u32,
        expected: Checksum,
        #[cfg_attr(
            not(feature = "page_ecc"),
            expect(unused_variables, reason = "recovery scheme only used under page_ecc")
        )]
        ecc_params: EccParams,
    ) -> crate::Result<(Slice, Option<EccRecoveryKind>)> {
        // Read straight into the Slice allocation — `read_exact` overwrites every
        // byte, so a zero-filled scratch buffer would be wasted, and returning a
        // Slice lets the no-compression / ECC-recovery callers avoid a later copy.
        let data = Slice::from_reader(reader, data_length as usize)?;

        let computed = Checksum::from_raw(crate::hash::hash128(&data));

        if ecc_length == 0 {
            computed.check(expected).inspect_err(|_| {
                log::error!(
                    "Checksum mismatch for block payload, got={computed}, expected={expected}",
                );
            })?;
            return Ok((data, None));
        }

        // ECC trailer present — always consume the parity bytes so
        // the reader cursor lands on the next block's header even
        // when the happy path doesn't need them.
        let parity = Slice::from_reader(reader, ecc_length as usize)?;

        if computed == expected {
            return Ok((data, None));
        }

        // Mismatch — try ECC recovery before failing.
        #[cfg(feature = "page_ecc")]
        {
            let expected_raw = expected.into_u128();

            // SEC-DED fast path: heal a single bit flip per word. The repaired
            // block must reproduce the stored checksum; a double-bit error (or
            // a heal that still mismatches) is surfaced, never silently
            // accepted. This scheme stores no shard parity, so there is no RS
            // fall-through here — the shard schemes are a separate EccParams
            // variant.
            if matches!(ecc_params, crate::table::block::EccParams::Secded) {
                // In-place single-bit heal needs an owned, mutable buffer; the
                // recovery path is rare, so the copy out of the read Slice is fine.
                let mut healed = data.to_vec();
                if crate::secded::try_correct_block(&mut healed, &parity)
                    == crate::secded::SecdedOutcome::Corrected
                    && crate::hash::hash128(&healed) == expected_raw
                {
                    log::warn!(
                        "recovered block via SEC-DED single-bit heal after \
                         checksum mismatch (data_len={}, ecc_len={ecc_length})",
                        data.len(),
                    );
                    return Ok((Slice::from(healed), Some(EccRecoveryKind::Secded)));
                }
                log::error!(
                    "Checksum mismatch on SEC-DED block, heal failed, \
                     got={computed}, expected={expected}",
                );
                return Err(crate::Error::PageEccUnrecoverable {
                    got: computed,
                    expected,
                });
            }

            let (data_shards, parity_shards) = ecc_params.as_shards();
            if let Some(recovered) = crate::ecc::try_recover(
                &data,
                &parity,
                data.len(),
                data_shards,
                parity_shards,
                |buf| crate::hash::hash128(buf) == expected_raw,
            )? {
                log::warn!(
                    "recovered block from RS parity after checksum mismatch \
                     (data_len={}, ecc_len={ecc_length})",
                    data.len(),
                );
                return Ok((Slice::from(recovered), Some(EccRecoveryKind::Shard)));
            }
            log::error!(
                "Checksum mismatch on ECC-protected block, recovery failed, \
                 got={computed}, expected={expected}",
            );
            Err(crate::Error::PageEccUnrecoverable {
                got: computed,
                expected,
            })
        }

        #[cfg(not(feature = "page_ecc"))]
        {
            // Block has an ECC trailer but this build can't use it.
            // Discard the parity buffer explicitly so the compiler
            // sees the use, then surface the checksum-mismatch
            // error directly. Earlier in this function we already
            // confirmed `computed != expected` (the `if computed
            // == expected` happy-path returned above), so
            // `computed.check(expected)` is guaranteed to return
            // Err here — return it directly instead of going
            // through `?` followed by an unreachable fallback.
            let _ = parity;
            log::error!(
                "block has ECC trailer (ecc_length={ecc_length}) but this \
                 build lacks the page_ecc feature — cannot attempt recovery; \
                 got={computed}, expected={expected}",
            );
            Err(crate::Error::ChecksumMismatch {
                expected,
                got: computed,
            })
        }
    }

    /// Encodes a block into a writer.
    ///
    /// Pipeline: raw data → compress → encrypt → checksum → write. The
    /// concrete pipeline shape (which steps run) is encoded by the
    /// [`BlockTransform`] variant (see its docs for the four valid
    /// combinations). The previous separate `(compression, encryption,
    /// zstd_dict)` argument triple has been collapsed into this single
    /// transform argument; `CompressionContext`'s constructors enforce
    /// that the dict bundle travels with the `ZstdDict` codec
    /// discriminator (see [`BlockTransform`] module docs), so the
    /// runtime `ZstdDictMismatch` guard inside this function is
    /// defensive only: every public construction path (direct
    /// `BlockTransform::Compressed(CompressionContext::with_dict(..))`
    /// and the [`BlockTransform::from_parts`] legacy helper) catches
    /// the mismatch before the call reaches `write_into`, so the
    /// guard is unreachable from any in-tree caller and exists purely
    /// as a "should-never-fire" assertion.
    pub fn write_into<W: crate::io::Write>(
        writer: &mut W,
        data: &[u8],
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
    ) -> crate::Result<Header> {
        // Most blocks carry no caller-supplied transform bits beyond what
        // the `transform` itself implies (compression / encryption / ECC).
        // The per-KV footer is the one bit `write_into` can't derive from
        // the payload, so the data-block writer routes through
        // `write_into_with_flags` to set it.
        Self::write_into_with_flags(writer, data, identity, transform, 0)
    }

    /// Like [`Self::write_into`] but lets the caller OR in
    /// transform-presence bits that aren't derivable from `transform`
    /// (currently only the `KV_CHECKSUM_FOOTER` flag from the `block_flags`
    /// module, since the footer lives in `data` and `write_into` can't see it).
    /// The compression / encryption / ECC bits are still derived from
    /// `transform` here, so the in-memory [`Header::block_flags`] always
    /// reflects the full transform stack regardless of which entry point is
    /// used. Note this is the IN-MEMORY header: only the self-describing block
    /// types (`Meta` / `Manifest` / `ManifestFooter`) serialize the
    /// `block_flags` byte to disk; SST block types omit it and the reader
    /// recovers transform presence from the per-SST meta descriptors, so a
    /// decoded SST header has `block_flags == 0`.
    ///
    /// Crate-internal: the `extra_flags` bag is a raw `u8` whose only valid
    /// bits live in the crate-private `block_flags` module, so an external
    /// caller could only guess magic values and a wrong bit would serialize
    /// a header claiming a transform the payload doesn't carry. External
    /// code uses the safe [`Self::write_into`] wrapper instead.
    pub(crate) fn write_into_with_flags<W: crate::io::Write>(
        writer: &mut W,
        data: &[u8],
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
        extra_flags: u8,
    ) -> crate::Result<Header> {
        Self::prepare_with_flags(data, identity, transform, extra_flags)?.write_to(writer)
    }

    /// Runs the block transform pipeline (compress → encrypt → checksum → ecc)
    /// and returns a [`PreparedBlock`] ready to be framed to disk by
    /// [`PreparedBlock::write_to`]. Pure CPU work, no I/O — safe to run on a
    /// worker thread for parallel compaction. See [`Self::write_into_with_flags`]
    /// for the `extra_flags` contract.
    #[expect(
        clippy::too_many_lines,
        reason = "linear transform pipeline: compress → encrypt → checksum → ecc; \
                  each step is small but they share state (header, payload, owned buffers) \
                  so factoring would just hide the data flow"
    )]
    pub(crate) fn prepare_with_flags<'a>(
        data: &'a [u8],
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
        extra_flags: u8,
    ) -> crate::Result<PreparedBlock<'a>> {
        // Unpack the transform back to the (compression, encryption,
        // zstd_dict) triple the implementation below was written
        // against. The transform's accessor methods carry the
        // pattern-match cost; the rest of the function keeps the same
        // shape as before the API collapse.
        let compression = transform.compression();
        let encryption = transform.encryption();
        #[cfg(zstd_any)]
        let zstd_dict = transform.zstd_dict();
        // Pull block_type out of identity so the rest of the
        // function reads exactly like the pre-identity version.
        // table_id / block_offset / dict_id / window_log are
        // accepted-but-not-consumed today — they'll feed AAD
        // construction once AEAD wiring lands. Call sites that
        // compute real values now won't need a second round of
        // edits when AAD goes live.
        let block_type = identity.block_type;
        if data.len() > MAX_DECOMPRESSION_SIZE as usize {
            return Err(crate::Error::DecompressedSizeTooLarge {
                declared: data.len() as u64,
                limit: u64::from(MAX_DECOMPRESSION_SIZE),
            });
        }

        // Self-describe the transform stack in the header. Compression /
        // encryption are derived from the active `transform`; the caller
        // ORs in any bit it can't derive (the per-KV footer). The ECC_PARITY
        // bit is NOT set here: it must agree with whether a parity trailer is
        // actually emitted, which is only known after the parity step below
        // (the ECC encoder short-circuits to a zero-length trailer for an
        // empty payload even when page_ecc is on). It is set after the parity
        // is computed, gated on a non-empty trailer, so the bit stays
        // presence-authoritative.
        let block_flags = {
            use crate::table::block::header::block_flags;
            // The header decoder rejects any bit outside `KNOWN`, so a caller
            // that ORs in a reserved bit here would write a block this build
            // can't read back. Catch that at the source in debug builds.
            debug_assert_eq!(
                extra_flags & !block_flags::KNOWN,
                0,
                "extra_flags must contain only defined block_flags bits",
            );
            let mut f = extra_flags;
            if transform.compression() != CompressionType::None {
                f |= block_flags::COMPRESSED;
            }
            if transform.encryption().is_some() {
                f |= block_flags::ENCRYPTED;
            }
            f
        };

        let mut header = Header {
            block_type,
            block_flags,
            checksum: Checksum::from_raw(0), // <-- NOTE: Is set later on
            data_length: 0,                  // <-- NOTE: Is set later on

            #[expect(clippy::cast_possible_truncation, reason = "blocks are limited to u32")]
            uncompressed_length: data.len() as u32,
        };

        // Compression step — produces an owned Vec when a compressor is active.
        #[cfg(any(feature = "lz4", zstd_any))]
        let mut compressed_buf: Option<Vec<u8>> = None;

        // Inner zstd-block layout, captured only for Data blocks compressed with
        // a non-dict zstd codec that split into >= 2 inner blocks (empty
        // otherwise). Enables range-query partial decode of large cold blocks.
        // `mut` is only exercised by the zstd Data arm below.
        #[cfg_attr(
            not(zstd_any),
            expect(unused_mut, reason = "`layout` is only mutated on zstd-enabled builds")
        )]
        let mut layout: Vec<u32> = Vec::new();

        match compression {
            CompressionType::None => {}

            #[cfg(feature = "lz4")]
            CompressionType::Lz4 => {
                compressed_buf = Some(lz4_flex::compress(data));
            }

            #[cfg(zstd_any)]
            CompressionType::Zstd(level) => {
                if block_type == BlockType::Data {
                    let (buf, lay) =
                        crate::compression::ZstdBackend::compress_with_layout(data, level)?;
                    compressed_buf = Some(buf);
                    layout = lay;
                } else {
                    compressed_buf = Some(crate::compression::ZstdBackend::compress(data, level)?);
                }
            }

            #[cfg(zstd_any)]
            CompressionType::ZstdDict { level, dict_id } => {
                let dict = zstd_dict.ok_or(crate::Error::ZstdDictMismatch {
                    expected: dict_id,
                    got: None,
                })?;
                if dict.id() != dict_id {
                    return Err(crate::Error::ZstdDictMismatch {
                        expected: dict_id,
                        got: Some(dict.id()),
                    });
                }

                compressed_buf = Some(crate::compression::ZstdBackend::compress_with_dict(
                    data,
                    level,
                    dict.raw(),
                )?);
            }
        }

        // Encryption step — under zstd this seals the AAD-bound envelope
        // binding the block identity + transform context; otherwise the opaque
        // form, reusing the owned compression buffer when present.
        let encrypted_buf: Option<Vec<u8>>;

        #[cfg(any(feature = "lz4", zstd_any))]
        {
            encrypted_buf = encryption
                .map(|enc| {
                    encrypt_block_payload(
                        enc,
                        compressed_buf.take(),
                        data,
                        &identity,
                        compression,
                        block_flags,
                    )
                })
                .transpose()?;
        }

        #[cfg(not(any(feature = "lz4", zstd_any)))]
        {
            encrypted_buf = encryption
                .map(|enc| {
                    encrypt_block_payload(enc, None, data, &identity, compression, block_flags)
                })
                .transpose()?;
        }

        // Determine the final on-disk payload. Owns a fresh buffer when a
        // transform produced one; borrows the caller's `data` otherwise, so the
        // serial uncompressed/unencrypted path stays zero-copy.
        let payload: Cow<'a, [u8]> = if let Some(enc) = encrypted_buf {
            Cow::Owned(enc)
        } else {
            #[cfg(any(feature = "lz4", zstd_any))]
            {
                compressed_buf.map_or(Cow::Borrowed(data), Cow::Owned)
            }
            #[cfg(not(any(feature = "lz4", zstd_any)))]
            {
                Cow::Borrowed(data)
            }
        };

        // Validate the final on-disk payload against the same size limit
        // enforced on the read path (MAX_DECOMPRESSION_SIZE + encryption overhead).
        // Check in u64 first to produce the correct DecompressedSizeTooLarge error,
        // then narrow to u32 for the header field.
        //
        // NOTE: max_overhead() is used only for the LIMIT — the actual ciphertext
        // length is checked against it regardless. A buggy provider that expands
        // beyond max_overhead() will be caught by this check (payload > limit).
        // Cap at u32::MAX to guarantee the subsequent as-u32 cast is safe.
        let max_payload = (u64::from(MAX_DECOMPRESSION_SIZE)
            + encryption.map_or(0u64, |enc| u64::from(enc.max_overhead())))
        .min(u64::from(u32::MAX));

        if payload.len() as u64 > max_payload {
            return Err(crate::Error::DecompressedSizeTooLarge {
                declared: payload.len() as u64,
                limit: max_payload,
            });
        }

        // Safe: payload.len() <= max_payload <= MAX_DECOMPRESSION_SIZE + overhead,
        // which is well within u32 range.
        #[expect(clippy::cast_possible_truncation, reason = "bounded by check above")]
        let payload_len = payload.len() as u32;

        header.data_length = payload_len;
        header.checksum = Checksum::from_raw(crate::hash::hash128(&payload));

        // Optional Reed-Solomon parity trailer. The parity LENGTH is not
        // stored in the header: it is `expected_parity_len(data_length)`,
        // recomputed by the reader from the presence-authoritative
        // `ECC_PARITY` block flag. On builds without the page_ecc feature,
        // transform.page_ecc() is a constant `false` (the Ecc variants of
        // BlockTransform don't exist), so the entire branch is dead and the
        // compiler folds it out.
        #[cfg(feature = "page_ecc")]
        let parity_buf: Option<Vec<u8>> = if let Some(ecc_params) = transform.ecc_params() {
            // SEC-DED emits one check byte per 8-byte word; shard schemes emit a
            // Reed-Solomon / XOR trailer. Both produce a parity buffer the
            // reader re-sizes from `data_length` via `expected_parity_len`.
            let p = match ecc_params {
                crate::table::block::EccParams::Secded => {
                    crate::secded::encode_block_parity(&payload)
                }
                crate::table::block::EccParams::Shard { .. } => {
                    let (data_shards, parity_shards) = ecc_params.as_shards();
                    crate::ecc::encode_parity(&payload, data_shards, parity_shards)?
                }
            };
            // parity_len is shard_bytes * RS_PARITY_SHARDS where
            // shard_bytes <= payload.len(). payload_len fits in u32
            // (checked above), so parity_len fits in u32 too; the
            // explicit try_from keeps the truncation path typed.
            let p_len =
                u32::try_from(p.len()).map_err(|_| crate::Error::DecompressedSizeTooLarge {
                    declared: p.len() as u64,
                    limit: u64::from(u32::MAX),
                })?;
            // Presence-authoritative ECC_PARITY bit: set only when a
            // non-empty parity trailer was actually emitted. An empty
            // payload yields a zero-length trailer (the encoder
            // short-circuits), so the bit stays clear and the reader
            // recomputes a zero parity length to match.
            if p_len > 0 {
                header.block_flags |= crate::table::block::header::block_flags::ECC_PARITY;
            }
            Some(p)
        } else {
            None
        };
        #[cfg(not(feature = "page_ecc"))]
        let parity_buf: Option<Vec<u8>> = None;

        Ok(PreparedBlock {
            header,
            payload,
            parity: parity_buf,
            layout,
        })
    }

    /// Reads a block from a reader.
    ///
    /// Pipeline: read → verify checksum → decrypt → decompress.
    /// When `encryption` is `None`, the decrypt step is skipped.
    ///
    /// Encryption state is determined by the caller (via [`Config`]),
    /// not recorded in the on-disk block header. With an authenticated
    /// encryption provider (such as AES-256-GCM), using the wrong key
    /// or provider will typically surface as a read/validation error
    /// (checksum, length, or decompression failure) rather than
    /// silently producing valid-looking plaintext.
    // The encrypted and unencrypted branches duplicate the checksum
    // verification and compression match because their input types
    // differ: encrypted reads into Vec<u8> (for decrypt_vec in-place
    // reuse), while unencrypted reads into Slice (zero-copy on the
    // None-compression path). Unifying them would require either a
    // Cow/enum wrapper or sacrificing the zero-copy optimization.
    #[expect(
        clippy::too_many_lines,
        reason = "encrypt/no-encrypt branches duplicate compression match — see comment above"
    )]
    pub fn from_reader<R: crate::io::Read>(
        reader: &mut R,
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
    ) -> crate::Result<Self> {
        let compression = transform.compression();
        let encryption = transform.encryption();
        #[cfg(zstd_any)]
        let zstd_dict = transform.zstd_dict();
        // `identity` (tree/table + dict/window context) feeds AAD
        // reconstruction on the encrypted path; `block_type` is still derived
        // from the parsed header (the frame self-describes it), not asserted
        // against identity.block_type.
        let header = Header::decode_from(reader)?;

        // Validate both size fields before any I/O or hashing to fail fast
        // on malformed headers. The on-disk data_length may include encryption
        // overhead (nonce + auth tag), so allow the provider's declared margin.
        // Use u64 arithmetic to avoid any possibility of u32 overflow
        // (consistent with from_file).
        let enc_overhead = encryption.map_or(0u64, |e| u64::from(e.max_overhead()));
        let max_data_length = u64::from(MAX_DECOMPRESSION_SIZE) + enc_overhead;

        if u64::from(header.data_length) > max_data_length {
            return Err(crate::Error::DecompressedSizeTooLarge {
                declared: u64::from(header.data_length),
                limit: max_data_length,
            });
        }

        if header.uncompressed_length > MAX_DECOMPRESSION_SIZE {
            return Err(crate::Error::DecompressedSizeTooLarge {
                declared: u64::from(header.uncompressed_length),
                limit: u64::from(MAX_DECOMPRESSION_SIZE),
            });
        }

        // Parity-trailer length is derived, not stored: when the block
        // carries a parity trailer (per `block_has_parity` — header bit for
        // self-describing blocks, descriptor-via-transform for SST blocks) it
        // is `expected_parity_len(data_length)` (the RS(4, 2) scheme is
        // deterministic), otherwise none.
        let ecc_length = if block_has_parity(&header, transform) {
            expected_parity_len(header.data_length, block_ecc_params(&header, transform))
        } else {
            0
        };

        // When encryption is active, read into a Vec so decrypt_vec can
        // reuse the buffer in-place (one allocation instead of two).
        // When no encryption, read into a Slice which may use optimized
        // reference-counted storage.
        let data = if let Some(enc) = encryption {
            // Read payload + optional ECC trailer, verify checksum
            // (with recovery on mismatch when parity is present).
            // `from_reader` returns no EccStatus, so a heal here is logged but
            // not surfaced; the status-returning `from_file_with_status` path is
            // what auto-heal observes.
            let (raw_vec, _recovery) = Self::read_payload_and_verify(
                reader,
                header.data_length,
                ecc_length,
                header.checksum,
                block_ecc_params(&header, transform),
            )?;

            // Recover the plaintext: AAD-bound envelope under zstd (verifies the
            // block identity), opaque in-place decrypt otherwise.
            let decrypted = decrypt_block_payload(enc, &raw_vec, &identity)?;

            match compression {
                CompressionType::None => {
                    #[expect(
                        clippy::cast_possible_truncation,
                        reason = "values are u32 length max"
                    )]
                    let actual_len = decrypted.len() as u32;

                    if header.uncompressed_length != actual_len {
                        return Err(crate::Error::InvalidHeader("Block"));
                    }

                    Slice::from(decrypted)
                }

                #[cfg(feature = "lz4")]
                CompressionType::Lz4 => {
                    // Decompress straight into the Slice's heap allocation,
                    // skipping both the zero-fill of `vec![0; n]` and the
                    // Vec -> Slice copy.
                    //
                    // SAFETY (load-bearing, do NOT reorder): the block checksum is
                    // verified ABOVE, before this point. That ordering is the
                    // precondition, not an optimization. `lz4_flex` is the
                    // unchecked fast decoder: it may wildcopy a back-reference out
                    // of the output buffer before detecting a malformed frame, so
                    // it must only ever run on a checksum-verified, intact stream.
                    // On such a stream every back-reference targets an
                    // already-written byte, so the uninitialized builder is written
                    // before it is ever read. Never move the decompress ahead of
                    // the checksum check.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut builder =
                        unsafe { Slice::builder_unzeroed(header.uncompressed_length as usize) };

                    let bytes_written = lz4_flex::decompress_into(&decrypted, &mut builder)
                        .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    builder.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::Zstd(_) => {
                    // Decompress straight into the Slice allocation: no
                    // zero-filled scratch Vec and no Vec -> Slice copy. SAFETY:
                    // same checksum-first precondition as the lz4 arm above. The
                    // stream is checksum-verified before this point, so the decoder
                    // only writes the uninitialized builder, never reads it
                    // unwritten. Do not reorder ahead of the checksum check.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut builder =
                        unsafe { Slice::builder_unzeroed(header.uncompressed_length as usize) };

                    let bytes_written =
                        crate::compression::ZstdBackend::decompress_into(&decrypted, &mut builder)
                            .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    builder.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::ZstdDict { dict_id, .. } => {
                    let dict = zstd_dict.ok_or(crate::Error::ZstdDictMismatch {
                        expected: dict_id,
                        got: None,
                    })?;
                    if dict.id() != dict_id {
                        return Err(crate::Error::ZstdDictMismatch {
                            expected: dict_id,
                            got: Some(dict.id()),
                        });
                    }

                    let decompressed = crate::compression::ZstdBackend::decompress_with_dict(
                        &decrypted,
                        dict,
                        header.uncompressed_length as usize,
                    )
                    .map_err(|_| crate::Error::Decompress(compression))?;

                    if decompressed.len() != header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    Slice::from(decompressed)
                }
            }
        } else {
            // Zero-copy fast path for non-ECC blocks (the v0..v5
            // legacy shape); ECC blocks go through the Vec-allocating
            // recovery-capable helper instead.
            let raw_data = if ecc_length == 0 {
                let s = Slice::from_reader(reader, header.data_length as usize)?;
                let checksum = Checksum::from_raw(crate::hash::hash128(&s));
                checksum.check(header.checksum).inspect_err(|_| {
                    log::error!(
                        "Checksum mismatch for <bufreader>, got={}, expected={}",
                        checksum,
                        header.checksum,
                    );
                })?;
                s
            } else {
                // from_reader has no EccStatus return; a heal is logged only.
                let (payload, _recovery) = Self::read_payload_and_verify(
                    reader,
                    header.data_length,
                    ecc_length,
                    header.checksum,
                    block_ecc_params(&header, transform),
                )?;
                payload
            };

            match compression {
                CompressionType::None => {
                    #[expect(
                        clippy::cast_possible_truncation,
                        reason = "values are u32 length max"
                    )]
                    let actual_len = raw_data.len() as u32;

                    if header.uncompressed_length != actual_len {
                        return Err(crate::Error::InvalidHeader("Block"));
                    }

                    raw_data
                }

                #[cfg(feature = "lz4")]
                CompressionType::Lz4 => {
                    // Decompress straight into the Slice's heap allocation,
                    // skipping the zero-fill and the Vec -> Slice copy. SAFETY:
                    // see the matching arm above — the checksum-verified stream
                    // is intact, so wildcopy never reads an unwritten position.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut builder =
                        unsafe { Slice::builder_unzeroed(header.uncompressed_length as usize) };

                    let bytes_written = lz4_flex::decompress_into(&raw_data, &mut builder)
                        .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    builder.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::Zstd(_) => {
                    // Decompress straight into the Slice allocation — no
                    // zero-filled scratch Vec and no Vec -> Slice copy.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut builder =
                        unsafe { Slice::builder_unzeroed(header.uncompressed_length as usize) };

                    let bytes_written =
                        crate::compression::ZstdBackend::decompress_into(&raw_data, &mut builder)
                            .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    builder.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::ZstdDict { dict_id, .. } => {
                    let dict = zstd_dict.ok_or(crate::Error::ZstdDictMismatch {
                        expected: dict_id,
                        got: None,
                    })?;
                    if dict.id() != dict_id {
                        return Err(crate::Error::ZstdDictMismatch {
                            expected: dict_id,
                            got: Some(dict.id()),
                        });
                    }

                    let decompressed = crate::compression::ZstdBackend::decompress_with_dict(
                        &raw_data,
                        dict,
                        header.uncompressed_length as usize,
                    )
                    .map_err(|_| crate::Error::Decompress(compression))?;

                    if decompressed.len() != header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    Slice::from(decompressed)
                }
            }
        };

        Ok(Self { header, data })
    }

    /// Reads a block from a file.
    ///
    /// Reads a block from a file, discarding the ECC status (warnings are
    /// still logged). Convenience wrapper over [`Self::from_file_with_status`]
    /// for the many call sites that don't act on the status.
    pub fn from_file(
        file: &dyn FsFile,
        handle: BlockHandle,
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
    ) -> crate::Result<Self> {
        let (block, _status) = Self::from_file_with_status(file, handle, identity, transform)?;
        Ok(block)
    }

    /// Reads a block from a file and reports its [`EccStatus`].
    ///
    /// Returns `Err` only on real payload corruption (checksum mismatch with
    /// no recovery available). On success, the [`EccStatus`] distinguishes a
    /// clean read ([`EccStatus::Ok`]) from one where the block carried an ECC
    /// trailer this build could not interpret ([`EccStatus::Unrecognized`] —
    /// the payload still verified by its checksum; recompaction re-stamps it
    /// with a supported scheme). A WARN is also logged in the latter case.
    ///
    /// Pipeline: read → verify checksum → decrypt → decompress. When
    /// `encryption` is `None`, the decrypt step is skipped.
    pub fn from_file_with_status(
        file: &dyn FsFile,
        handle: BlockHandle,
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
    ) -> crate::Result<(Self, EccStatus)> {
        let (block, status, _recovery) =
            Self::from_file_with_recovery(file, handle, identity, transform)?;
        Ok((block, status))
    }

    /// Like [`Self::from_file_with_status`] but additionally reports which ECC
    /// mechanism repaired the block (`Some(kind)` iff the status is
    /// [`EccStatus::Corrected`]). Internal to the read path: the primary read
    /// call sites (`load_block`, the partial-decode path, patrol scrub) use it
    /// to attribute the on-read recovery to the right metric counter, keeping
    /// the public [`EccStatus`] free of the kind.
    // Same duplication rationale as from_reader — see comment there.
    #[expect(
        clippy::too_many_lines,
        reason = "encrypt/no-encrypt branches duplicate compression match — see from_reader"
    )]
    pub(crate) fn from_file_with_recovery(
        file: &dyn FsFile,
        handle: BlockHandle,
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
    ) -> crate::Result<(Self, EccStatus, Option<EccRecoveryKind>)> {
        let compression = transform.compression();
        let encryption = transform.encryption();
        #[cfg(zstd_any)]
        let zstd_dict = transform.zstd_dict();
        // `identity` (tree/table + compression context) feeds AAD
        // reconstruction on the encrypted read path below.
        // handle.size() includes Header::MIN_LEN + payload +
        // optional ECC parity trailer. Encrypted blocks add
        // provider-specific overhead to the on-disk size, AND ECC
        // parity scales with the (encrypted) payload — about
        // (data_length + enc_overhead) / 2 + 4 bytes.
        //
        // Sum of parts: header + max_payload + parity(max_payload)
        // where max_payload = MAX_DECOMPRESSION_SIZE + enc_overhead.
        // A MAX_DECOMPRESSION_SIZE-only ECC bound would
        // under-approximate by ~enc_overhead/2 and reject legitimate
        // near-limit encrypted+ECC blocks the writer can produce.
        let enc_overhead = encryption.map_or(0u64, |e| u64::from(e.max_overhead()));
        let max_payload = u64::from(MAX_DECOMPRESSION_SIZE) + enc_overhead;
        // Pre-allocation sanity cap on `handle.size()`: reject an absurd on-disk
        // size before allocating the read buffer. The ECC-OFF path adds NO ECC
        // term and runs NO ECC math — when there is no parity, the cap is just
        // payload + header. When ECC is on, the cap allows the block's ACTUAL
        // scheme parity (from the per-SST descriptor carried by `transform`),
        // never a hardcoded scheme. Self-describing blocks (Meta / Manifest)
        // carry their own small RS parity but are orders of magnitude below
        // `max_payload`, so they pass this cap without an explicit ECC term.
        let max_ecc_overhead = match transform.ecc_params() {
            Some(params) => {
                #[expect(
                    clippy::cast_possible_truncation,
                    reason = "max_payload is MAX_DECOMPRESSION_SIZE (+ enc overhead), well below u32::MAX"
                )]
                let max_payload_u32 = max_payload.min(u64::from(u32::MAX)) as u32;
                u64::from(expected_parity_len(max_payload_u32, params))
            }
            None => 0,
        };
        let max_on_disk_size = max_payload + max_ecc_overhead + Header::MAX_LEN as u64;

        if u64::from(handle.size()) > max_on_disk_size {
            return Err(crate::Error::DecompressedSizeTooLarge {
                declared: u64::from(handle.size()),
                limit: max_on_disk_size,
            });
        }

        // When encryption is active, read the whole block into an owned
        // Vec (single I/O, single allocation), parse the header, then strip
        // the header prefix so decrypt_vec operates on the payload in-place.
        // No intermediate Slice, no overlap of encrypted + decrypted buffers.
        // When no encryption, read into a Slice (zero-copy on the
        // None-compression path).
        let (header, data, ecc_status, recovery) = if let Some(enc) = encryption {
            let block_size = handle.size() as usize;

            // Pre-decode lower bound: every header is at least MIN_LEN; the
            // exact length (with or without the block_flags byte) is known
            // only after the block_type is decoded.
            if block_size < Header::MIN_LEN {
                return Err(crate::Error::InvalidHeader("Block"));
            }

            // Zero-init is redundant (read_at overwrites all bytes) but avoids
            // unsafe. The cost is negligible vs I/O + decryption. Unsafe
            // uninitialized allocation (like Slice::builder_unzeroed) could be
            // used here if profiling shows this as a bottleneck.
            let mut buf = vec![0u8; block_size];
            let n = file.read_at(&mut buf, *handle.offset())?;
            if n != block_size {
                return Err(crate::Error::Io(crate::io::Error::new(
                    crate::io::ErrorKind::UnexpectedEof,
                    format!(
                        "block read_at: expected {block_size} bytes, got {n} at offset {}",
                        *handle.offset(),
                    ),
                )));
            }

            // `decode_from` reads exactly the header (variable: 33 or 34
            // bytes per block_type) and stops, leaving the payload untouched.
            let parsed_header = Header::decode_from(&mut &buf[..])?;
            let header_len = Header::header_len(parsed_header.block_type);

            // Parity-trailer presence is keyed on a RECOGNIZED ECC layout
            // (`block_has_parity`: header bit for self-describing blocks,
            // descriptor-via-transform for SST), NOT on `ecc_length` (which is
            // also 0 for a recognized scheme on an empty payload). The trailer
            // length, when recognized, is derived from `data_length` + scheme.
            let has_ecc = block_has_parity(&parsed_header, transform);
            let ecc_length = if has_ecc {
                expected_parity_len(
                    parsed_header.data_length,
                    block_ecc_params(&parsed_header, transform),
                )
            } else {
                0
            };

            // Clamp-to-zero: a block truncated before its header ends has no
            // payload, which `classify_block_trailer` then flags as a mismatch.
            let actual_payload_plus_ecc = block_size.saturating_sub(header_len);
            let actual_data_len = parsed_header.data_length as usize;
            let ecc_status = classify_block_trailer(
                has_ecc,
                actual_payload_plus_ecc,
                actual_data_len,
                ecc_length,
                &handle,
            )?;

            // Payload-length safety cap. Mirrors the `from_reader`
            // check (see `Block::from_reader` above): the on-disk
            // size cap on `handle.size()` allows for ECC parity
            // overhead, so a malformed non-ECC block could declare
            // `data_length` ≈ `MAX_DECOMPRESSION_SIZE * 1.5`
            // (ECC-inclusive bound) and pass the outer check.
            // Reject those explicitly here, before any further work
            // trusts the declared payload length.
            let max_data_length = u64::from(MAX_DECOMPRESSION_SIZE) + enc_overhead;
            if u64::from(parsed_header.data_length) > max_data_length {
                return Err(crate::Error::DecompressedSizeTooLarge {
                    declared: u64::from(parsed_header.data_length),
                    limit: max_data_length,
                });
            }

            if parsed_header.uncompressed_length > MAX_DECOMPRESSION_SIZE {
                return Err(crate::Error::DecompressedSizeTooLarge {
                    declared: u64::from(parsed_header.uncompressed_length),
                    limit: u64::from(MAX_DECOMPRESSION_SIZE),
                });
            }

            // ECC fast path: no recognized parity trailer → in-buffer
            // checksum check + decrypt_vec. The checksum covers exactly the
            // `data_length` payload bytes (NOT any trailing bytes), so an
            // unrecognized opaque trailer is excluded and discarded. With a
            // recognized scheme, run the shared helper against a cursor over
            // the post-header bytes so recovery is available on mismatch.
            let (buf, payload_corrected) = if ecc_length == 0 {
                #[expect(
                    clippy::indexing_slicing,
                    reason = "actual_data_len <= post-header len"
                )]
                let checksum = Checksum::from_raw(crate::hash::hash128(
                    &buf[header_len..header_len + actual_data_len],
                ));
                checksum.check(parsed_header.checksum).inspect_err(|_| {
                    log::error!(
                        "Checksum mismatch for block {handle:?}, got={}, expected={}",
                        checksum,
                        parsed_header.checksum,
                    );
                })?;
                // Strip header prefix + any opaque trailer so buf is the payload.
                buf.copy_within(header_len..header_len + actual_data_len, 0);
                buf.truncate(actual_data_len);
                (Slice::from(buf), None)
            } else {
                #[expect(clippy::indexing_slicing, reason = "header was decoded from buf")]
                let mut cursor = crate::io::Cursor::new(&buf[header_len..]);
                Self::read_payload_and_verify(
                    &mut cursor,
                    parsed_header.data_length,
                    ecc_length,
                    parsed_header.checksum,
                    block_ecc_params(&parsed_header, transform),
                )?
            };

            // Fold a successful ECC repair into the reported status; an
            // unrecognized scheme never heals, so the two are exclusive. The
            // recovery mechanism is carried out separately as `payload_corrected`.
            let ecc_status = if payload_corrected.is_some() {
                EccStatus::Corrected
            } else {
                ecc_status
            };

            let decrypted = decrypt_block_payload(enc, &buf, &identity)?;

            let data = match compression {
                CompressionType::None => {
                    #[expect(
                        clippy::cast_possible_truncation,
                        reason = "values are u32 length max"
                    )]
                    let actual_len = decrypted.len() as u32;

                    if parsed_header.uncompressed_length != actual_len {
                        return Err(crate::Error::InvalidHeader("Block"));
                    }

                    Slice::from(decrypted)
                }

                #[cfg(feature = "lz4")]
                CompressionType::Lz4 => {
                    // Decompress straight into the Slice allocation (no zero-fill,
                    // no Vec -> Slice copy). SAFETY: see the read-path arm above;
                    // the checksum-verified stream is intact.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut decompressed = unsafe {
                        Slice::builder_unzeroed(parsed_header.uncompressed_length as usize)
                    };

                    let bytes_written = lz4_flex::decompress_into(&decrypted, &mut decompressed)
                        .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != parsed_header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    decompressed.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::Zstd(_) => {
                    // Decompress straight into the Slice allocation — no
                    // zero-filled scratch Vec and no Vec -> Slice copy.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut builder = unsafe {
                        Slice::builder_unzeroed(parsed_header.uncompressed_length as usize)
                    };

                    let bytes_written =
                        crate::compression::ZstdBackend::decompress_into(&decrypted, &mut builder)
                            .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != parsed_header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    builder.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::ZstdDict { dict_id, .. } => {
                    let dict = zstd_dict.ok_or(crate::Error::ZstdDictMismatch {
                        expected: dict_id,
                        got: None,
                    })?;
                    if dict.id() != dict_id {
                        return Err(crate::Error::ZstdDictMismatch {
                            expected: dict_id,
                            got: Some(dict.id()),
                        });
                    }

                    let decompressed = crate::compression::ZstdBackend::decompress_with_dict(
                        &decrypted,
                        dict,
                        parsed_header.uncompressed_length as usize,
                    )
                    .map_err(|_| crate::Error::Decompress(compression))?;

                    if decompressed.len() != parsed_header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    Slice::from(decompressed)
                }
            };

            (parsed_header, data, ecc_status, payload_corrected)
        } else {
            // Single I/O read — header + payload in one Slice.
            let buf = crate::file::read_exact(file, *handle.offset(), handle.size() as usize)?;

            let parsed_header = Header::decode_from(&mut &buf[..])?;
            let header_len = Header::header_len(parsed_header.block_type);

            // Recognized-ECC presence keys on `block_has_parity`, not on
            // `ecc_length` (which is also 0 for a recognized scheme on an empty
            // payload). See the encrypted branch + `classify_block_trailer`.
            let has_ecc = block_has_parity(&parsed_header, transform);
            let ecc_length = if has_ecc {
                expected_parity_len(
                    parsed_header.data_length,
                    block_ecc_params(&parsed_header, transform),
                )
            } else {
                0
            };

            // Clamp-to-zero: a buffer shorter than the header carries no payload,
            // which the trailer classification then flags as a mismatch.
            let actual_payload_plus_ecc = buf.len().saturating_sub(header_len);
            let actual_data_len = parsed_header.data_length as usize;
            let ecc_status = classify_block_trailer(
                has_ecc,
                actual_payload_plus_ecc,
                actual_data_len,
                ecc_length,
                &handle,
            )?;

            if parsed_header.uncompressed_length > MAX_DECOMPRESSION_SIZE {
                return Err(crate::Error::DecompressedSizeTooLarge {
                    declared: u64::from(parsed_header.uncompressed_length),
                    limit: u64::from(MAX_DECOMPRESSION_SIZE),
                });
            }

            // Zero-copy fast path for non-recovery blocks (Off or unrecognized
            // opaque trailer); recognized-ECC blocks go through the
            // recovery-capable helper. The checksum covers exactly the
            // `data_length` payload bytes, so an opaque trailer is excluded.
            let (payload_slice, payload_corrected): (Slice, Option<EccRecoveryKind>) =
                if ecc_length == 0 {
                    #[expect(
                        clippy::indexing_slicing,
                        reason = "actual_data_len <= post-header len"
                    )]
                    let checksum = Checksum::from_raw(crate::hash::hash128(
                        &buf[header_len..header_len + actual_data_len],
                    ));
                    checksum.check(parsed_header.checksum).inspect_err(|_| {
                        log::error!(
                            "Checksum mismatch for block {handle:?}, got={}, expected={}",
                            checksum,
                            parsed_header.checksum,
                        );
                    })?;
                    (buf.slice(header_len..header_len + actual_data_len), None)
                } else {
                    #[expect(clippy::indexing_slicing, reason = "header was decoded from buf")]
                    let mut cursor = crate::io::Cursor::new(&buf[header_len..]);
                    let (payload, recovery) = Self::read_payload_and_verify(
                        &mut cursor,
                        parsed_header.data_length,
                        ecc_length,
                        parsed_header.checksum,
                        block_ecc_params(&parsed_header, transform),
                    )?;
                    (payload, recovery)
                };
            // Fold a successful ECC repair into the status; the recovery
            // mechanism is carried out separately as `payload_corrected`.
            let ecc_status = if payload_corrected.is_some() {
                EccStatus::Corrected
            } else {
                ecc_status
            };

            let data = match compression {
                CompressionType::None => {
                    #[expect(
                        clippy::cast_possible_truncation,
                        reason = "values are u32 length max"
                    )]
                    let actual_len = payload_slice.len() as u32;

                    if parsed_header.uncompressed_length != actual_len {
                        return Err(crate::Error::InvalidHeader("Block"));
                    }

                    payload_slice
                }

                #[cfg(feature = "lz4")]
                CompressionType::Lz4 => {
                    let compressed_data: &[u8] = &payload_slice;

                    // Decompress straight into the Slice allocation (no zero-fill,
                    // no Vec -> Slice copy). SAFETY: see the read-path arm above;
                    // the checksum-verified stream is intact.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut decompressed = unsafe {
                        Slice::builder_unzeroed(parsed_header.uncompressed_length as usize)
                    };

                    let bytes_written =
                        lz4_flex::decompress_into(compressed_data, &mut decompressed)
                            .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != parsed_header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    decompressed.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::Zstd(_) => {
                    let compressed_data: &[u8] = &payload_slice;

                    // Decompress straight into the Slice allocation — no
                    // zero-filled scratch Vec and no Vec -> Slice copy.
                    #[expect(unsafe_code, reason = "fill an uninitialized Slice via decompress")]
                    let mut builder = unsafe {
                        Slice::builder_unzeroed(parsed_header.uncompressed_length as usize)
                    };

                    let bytes_written = crate::compression::ZstdBackend::decompress_into(
                        compressed_data,
                        &mut builder,
                    )
                    .map_err(|_| crate::Error::Decompress(compression))?;

                    if bytes_written != parsed_header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    builder.freeze().into()
                }

                #[cfg(zstd_any)]
                CompressionType::ZstdDict { dict_id, .. } => {
                    let compressed_data: &[u8] = &payload_slice;

                    let dict = zstd_dict.ok_or(crate::Error::ZstdDictMismatch {
                        expected: dict_id,
                        got: None,
                    })?;
                    if dict.id() != dict_id {
                        return Err(crate::Error::ZstdDictMismatch {
                            expected: dict_id,
                            got: Some(dict.id()),
                        });
                    }

                    let decompressed = crate::compression::ZstdBackend::decompress_with_dict(
                        compressed_data,
                        dict,
                        parsed_header.uncompressed_length as usize,
                    )
                    .map_err(|_| crate::Error::Decompress(compression))?;

                    if decompressed.len() != parsed_header.uncompressed_length as usize {
                        return Err(crate::Error::Decompress(compression));
                    }

                    Slice::from(decompressed)
                }
            };

            (parsed_header, data, ecc_status, payload_corrected)
        };

        Ok((Self { header, data }, ecc_status, recovery))
    }

    /// Reads a data block's verified COMPRESSED payload (the zstd frame) WITHOUT
    /// decompressing it, for partial / lazy decode. Returns the header, the
    /// compressed-frame bytes (checksum-verified; ECC-recovered if a recognized
    /// parity trailer is present), and `Some(kind)` when a recovery occurred so
    /// the caller can schedule auto-heal and count the recovery.
    ///
    /// Non-encrypted blocks only — the caller must ensure
    /// `transform.encryption().is_none()` (an encrypted block's plaintext frame
    /// is only available after a whole-block decrypt, which defeats the lazy
    /// path). Mirrors the non-encrypted read+verify of
    /// [`Self::from_file_with_status`], stopping before decompression.
    ///
    /// # Errors
    ///
    /// Returns an error on a framing / checksum failure (unrecoverable), or if
    /// called for an encrypted transform.
    #[cfg(feature = "zstd")]
    pub(crate) fn read_data_frame(
        file: &dyn FsFile,
        handle: BlockHandle,
        transform: &BlockTransform<'_>,
    ) -> crate::Result<(Header, Slice, Option<EccRecoveryKind>)> {
        if transform.encryption().is_some() {
            return Err(crate::Error::Io(crate::io::Error::other(
                "read_data_frame: encrypted blocks are not supported on the lazy path",
            )));
        }

        // Pre-allocation sanity cap on `handle.size()`, mirroring
        // `from_file_with_status`: reject an absurd on-disk size before
        // allocating the read buffer, so a corrupt handle cannot force a huge
        // allocation. Non-encrypted path (encryption rejected above), so no
        // encryption overhead; the ECC term uses the block's ACTUAL per-SST
        // scheme (never a hardcoded one) and is 0 when there is no parity.
        let max_ecc_overhead = match transform.ecc_params() {
            Some(params) => u64::from(expected_parity_len(MAX_DECOMPRESSION_SIZE, params)),
            None => 0,
        };
        let max_on_disk_size =
            u64::from(MAX_DECOMPRESSION_SIZE) + max_ecc_overhead + Header::MAX_LEN as u64;
        if u64::from(handle.size()) > max_on_disk_size {
            return Err(crate::Error::DecompressedSizeTooLarge {
                declared: u64::from(handle.size()),
                limit: max_on_disk_size,
            });
        }

        let buf = crate::file::read_exact(file, *handle.offset(), handle.size() as usize)?;
        let parsed_header = Header::decode_from(&mut &buf[..])?;
        // Reject a header that declares a decompressed size past the cap before
        // the lazy decode path trusts it.
        if parsed_header.uncompressed_length > MAX_DECOMPRESSION_SIZE {
            return Err(crate::Error::DecompressedSizeTooLarge {
                declared: u64::from(parsed_header.uncompressed_length),
                limit: u64::from(MAX_DECOMPRESSION_SIZE),
            });
        }
        let header_len = Header::header_len(parsed_header.block_type);

        let has_ecc = block_has_parity(&parsed_header, transform);
        let ecc_length = if has_ecc {
            expected_parity_len(
                parsed_header.data_length,
                block_ecc_params(&parsed_header, transform),
            )
        } else {
            0
        };

        // Clamp-to-zero: a buffer shorter than the header carries no payload,
        // which the trailer classification then flags as a mismatch.
        let actual_payload_plus_ecc = buf.len().saturating_sub(header_len);
        let actual_data_len = parsed_header.data_length as usize;
        let _ecc_status = classify_block_trailer(
            has_ecc,
            actual_payload_plus_ecc,
            actual_data_len,
            ecc_length,
            &handle,
        )?;

        let (payload, recovery): (Slice, Option<EccRecoveryKind>) = if ecc_length == 0 {
            #[expect(
                clippy::indexing_slicing,
                reason = "actual_data_len <= post-header len, checked via classify_block_trailer"
            )]
            let checksum = Checksum::from_raw(crate::hash::hash128(
                &buf[header_len..header_len + actual_data_len],
            ));
            checksum.check(parsed_header.checksum)?;
            (buf.slice(header_len..header_len + actual_data_len), None)
        } else {
            #[expect(clippy::indexing_slicing, reason = "header was decoded from buf")]
            let mut cursor = crate::io::Cursor::new(&buf[header_len..]);
            // `recovery` is surfaced so the partial-decode caller can schedule
            // auto-heal (the recovered bytes are correct but the on-disk copy is
            // still faulty) and count the recovery; a heal is also logged inside
            // read_payload_and_verify.
            let (frame, recovery) = Self::read_payload_and_verify(
                &mut cursor,
                parsed_header.data_length,
                ecc_length,
                parsed_header.checksum,
                block_ecc_params(&parsed_header, transform),
            )?;
            (frame, recovery)
        };

        Ok((parsed_header, payload, recovery))
    }
}

#[cfg(test)]
#[allow(
    clippy::unwrap_used,
    clippy::indexing_slicing,
    clippy::useless_vec,
    clippy::cast_possible_truncation,
    clippy::expect_used,
    reason = "test code"
)]
mod tests {
    use super::*;
    use test_log::test;

    /// A pathological-but-valid shard config (1 data shard, 255 parity shards)
    /// over a large block makes `shard_bytes * parity_shards` exceed u32. The
    /// parity length must saturate to `u32::MAX` (it is rejected against the actual
    /// block downstream), never wrap or panic.
    #[test]
    fn expected_parity_len_saturates_on_huge_parity_product() {
        let data_length = 100 * 1024 * 1024; // 100 MiB
        let params = EccParams::Shard {
            data_shards: 1,
            parity_shards: 255,
        };
        assert_eq!(expected_parity_len(data_length, params), u32::MAX);
    }

    /// A `u32::MAX` data length with a single data shard makes `ceil` equal an
    /// odd `u32::MAX`; rounding it up to an even shard size must saturate, not
    /// overflow `u32`. A corrupt header reaching this path is rejected
    /// downstream, so the clamp must not panic here.
    #[test]
    fn expected_parity_len_saturates_on_max_data_length_even_rounding() {
        let params = EccParams::Shard {
            data_shards: 1,
            parity_shards: 2,
        };
        assert_eq!(expected_parity_len(u32::MAX, params), u32::MAX);
    }

    /// Result of [`write_block_to_tempfile`]. Bundles the open
    /// file, the pre-computed [`crate::table::BlockHandle`], and
    /// the owning [`tempfile::TempDir`].
    ///
    /// **Drop-order safety lives entirely in the struct field
    /// order** (Windows portability constraint). Rust drops
    /// struct fields in declaration order, so `file` is first
    /// and `dir` LAST: when a `TempBlock` value goes out of
    /// scope, the open file handle closes before the `TempDir`
    /// removes the directory. Windows rejects directory
    /// removal while a file inside it is still open.
    ///
    /// **Callers SHOULD keep the result as a single binding**
    /// — borrow `&tmp.file` and copy `tmp.handle` (it's `Copy`)
    /// instead of destructuring. Destructuring opens a
    /// foot-gun: local bindings drop in REVERSE declaration
    /// order, so a pattern like
    /// `let TempBlock { file, handle, dir: _dir } = ...?;`
    /// would close `dir` before `file` and break the
    /// invariant. Holding the whole struct as one local
    /// (`let tmp = ...?;`) makes the struct field order the
    /// SOLE source of truth — no pattern can break it.
    struct TempBlock {
        /// Open read-only handle on the persisted block.
        /// Declared first so it drops before `dir`.
        file: std::fs::File,
        /// Pre-computed handle: offset 0, length = header +
        /// payload, ready to pass straight into `Block::from_file`.
        handle: crate::table::BlockHandle,
        /// Drop guard for the tempdir. Kept bound for the test
        /// lifetime so the file inside the directory survives
        /// long enough to be reopened by the test reader, then
        /// dropped at end-of-scope to reclaim the directory.
        /// Declared LAST so the file handle above closes before
        /// this removes the directory.
        ///
        /// Feature-matrix-gated suppression: under `--all-features`
        /// the ECC-corruption tests read `.path()` to flip on-disk
        /// bytes, so the field IS used and an unconditional
        /// `#[allow(dead_code)]` would clash with
        /// `clippy::used_underscore_binding` if we tried to prefix
        /// the field name with `_`. Under default features the
        /// `page_ecc` tests are not compiled, the field genuinely
        /// IS dead, and the suppression silences the warning.
        #[cfg_attr(
            not(feature = "page_ecc"),
            expect(dead_code, reason = "drop guard; only read by page_ecc-gated tests")
        )]
        dir: tempfile::TempDir,
    }

    /// Shared scaffold for `Block::from_file` roundtrip tests: writes
    /// `data` through `Block::write_into` directly into a fresh
    /// tempdir-backed file, reopens it read-only, and returns a
    /// [`TempBlock`] bundling the open file, the pre-computed
    /// [`crate::table::BlockHandle`], and the owning [`tempfile::TempDir`]
    /// (kept bound for the test's lifetime). The streaming write
    /// avoids an intermediate `Vec<u8>` — relevant for the 32 KiB
    /// large-payload encryption test.
    ///
    /// Centralises the ~10× write/sync/reopen/handle boilerplate that
    /// the `from_file` tests below would otherwise duplicate.
    fn write_block_to_tempfile(
        data: &[u8],
        identity: BlockIdentity,
        transform: &BlockTransform<'_>,
    ) -> crate::Result<TempBlock> {
        let dir = tempfile::tempdir()?;
        let path = dir.path().join("block");
        // Scope the write-side file handle so the read-side
        // `File::open` below sees a fully-flushed file. Dropping
        // closes; sync_all flushes before close.
        let header = {
            let mut file = std::fs::File::create(&path)?;
            let header = Block::write_into(&mut file, data, identity, transform)?;
            file.sync_all()?;
            header
        };
        let file = std::fs::File::open(&path)?;
        // Size the handle under the transform's actual scheme so the helper
        // works for any ECC layout, not just RS(4,2). `on_disk_size()` hardcodes
        // RS(4,2) and would mis-size a block written with a different scheme.
        let handle = crate::table::BlockHandle::new(
            BlockOffset(0),
            header.on_disk_size_with(transform.ecc_params()),
        );
        Ok(TempBlock { file, handle, dir })
    }

    #[test]
    fn block_from_file_roundtrip_uncompressed() -> crate::Result<()> {
        let data = b"abcdefabcdefabcdef";
        let tmp = write_block_to_tempfile(
            data,
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::None,
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;
        let block = Block::from_file(
            &tmp.file,
            tmp.handle,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::None,
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;
        assert_eq!(data, &*block.data);
        Ok(())
    }

    #[cfg(feature = "zstd")]
    #[test]
    fn read_data_frame_returns_decompressible_zstd_frame() -> crate::Result<()> {
        // Repetitive payload so zstd produces a real (smaller) compressed frame.
        let data: Vec<u8> = (0..40_000u32).map(|i| (i % 64) as u8).collect();
        let transform =
            crate::table::block::BlockTransform::from_parts(CompressionType::Zstd(3), None, None)?;
        let tmp = write_block_to_tempfile(
            &data,
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &transform,
        )?;

        let (header, frame, _corrected) =
            Block::read_data_frame(&tmp.file, tmp.handle, &transform)?;
        // The returned bytes are the COMPRESSED frame: it must be smaller than
        // the payload and must decompress back to the original (proving
        // read_data_frame skipped decode yet returned a valid frame).
        assert!(
            frame.len() < data.len(),
            "frame must be compressed (got {} for {} bytes)",
            frame.len(),
            data.len(),
        );
        let decompressed = crate::compression::ZstdBackend::decompress(
            &frame,
            header.uncompressed_length as usize + 1,
        )?;
        assert_eq!(decompressed, data, "frame must decompress to the original");
        Ok(())
    }

    #[cfg(feature = "zstd")]
    #[test]
    fn read_data_frame_rejects_oversized_handle() -> crate::Result<()> {
        // A handle declaring an absurd on-disk size must be rejected by the
        // pre-allocation cap before any read / allocation, mirroring
        // `from_file_with_status`.
        let data: Vec<u8> = (0..1_000u32).map(|i| (i % 64) as u8).collect();
        let transform =
            crate::table::block::BlockTransform::from_parts(CompressionType::Zstd(3), None, None)?;
        let tmp = write_block_to_tempfile(
            &data,
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &transform,
        )?;
        let oversized = BlockHandle::new(tmp.handle.offset(), u32::MAX);
        let err = Block::read_data_frame(&tmp.file, oversized, &transform)
            .expect_err("oversized handle must be rejected");
        assert!(
            matches!(err, crate::Error::DecompressedSizeTooLarge { .. }),
            "expected DecompressedSizeTooLarge, got {err:?}",
        );
        Ok(())
    }

    #[test]
    #[cfg(feature = "lz4")]
    fn block_from_file_roundtrip_lz4() -> crate::Result<()> {
        let data = b"abcdefabcdefabcdef";
        let tmp = write_block_to_tempfile(
            data,
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;
        let block = Block::from_file(
            &tmp.file,
            tmp.handle,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;
        assert_eq!(data, &*block.data);
        Ok(())
    }

    #[test]
    #[cfg(zstd_any)]
    fn block_from_file_roundtrip_zstd() -> crate::Result<()> {
        let data = b"abcdefabcdefabcdef";
        let tmp = write_block_to_tempfile(
            data,
            BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Zstd(3),
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;
        let block = Block::from_file(
            &tmp.file,
            tmp.handle,
            BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Zstd(3),
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;
        assert_eq!(data, &*block.data);
        Ok(())
    }

    #[test]
    fn block_roundtrip_uncompressed() -> crate::Result<()> {
        let mut writer = vec![];

        Block::write_into(
            &mut writer,
            b"abcdefabcdefabcdef",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::None,
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;

        {
            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    None,
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(b"abcdefabcdefabcdef", &*block.data);
        }

        Ok(())
    }
    #[test]
    #[cfg(feature = "lz4")]
    fn block_roundtrip_lz4() -> crate::Result<()> {
        let mut writer = vec![];

        Block::write_into(
            &mut writer,
            b"abcdefabcdefabcdef",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;

        {
            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Lz4,
                    None,
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(b"abcdefabcdefabcdef", &*block.data);
        }

        Ok(())
    }

    #[test]
    #[cfg(feature = "lz4")]
    fn block_reject_absurd_uncompressed_length() {
        use crate::coding::Encode;

        // Write a valid lz4-compressed block first so we get the right header format
        let mut buf = vec![];
        Block::write_into(
            &mut buf,
            b"hello",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        )
        .unwrap();

        // Tamper the header: set uncompressed_length to u32::MAX.
        // The block checksum only covers the compressed payload bytes; it does not include
        // header fields. The header itself has its own checksum, which we recompute below
        // by re-encoding the modified header, so the tampered block remains internally
        // consistent while exercising the DecompressedSizeTooLarge path.
        let mut reader = &buf[..];
        let mut header = Header::decode_from(&mut reader).unwrap();
        let compressed_payload: Vec<u8> = reader.to_vec();

        header.uncompressed_length = u32::MAX;
        let mut tampered = header.encode_into_vec();
        tampered.extend_from_slice(&compressed_payload);

        let mut r = &tampered[..];
        let result = Block::from_reader(
            &mut r,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        assert!(
            matches!(&result, Err(crate::Error::DecompressedSizeTooLarge { .. })),
            "expected DecompressedSizeTooLarge, got: {:?}",
            result.err(),
        );
    }

    #[test]
    #[cfg(feature = "lz4")]
    fn block_zero_uncompressed_length_with_data_fails_decompress() {
        use crate::coding::Encode;

        // Zero uncompressed_length is allowed (valid for empty blocks), but when
        // the compressed payload is non-empty, lz4 decompression will fail because
        // the output buffer is zero-sized.
        let mut buf = vec![];
        Block::write_into(
            &mut buf,
            b"hello",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        )
        .unwrap();

        let mut reader = &buf[..];
        let mut header = Header::decode_from(&mut reader).unwrap();
        let compressed_payload: Vec<u8> = reader.to_vec();

        header.uncompressed_length = 0;
        let mut tampered = header.encode_into_vec();
        tampered.extend_from_slice(&compressed_payload);

        let mut r = &tampered[..];
        let result = Block::from_reader(
            &mut r,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        assert!(
            matches!(&result, Err(crate::Error::Decompress(_))),
            "expected Decompress error, got: {:?}",
            result.err(),
        );
    }

    #[test]
    #[cfg(feature = "lz4")]
    fn lz4_corrupted_uncompressed_length_triggers_decompress_error() {
        use crate::coding::Encode;
        use std::io::Cursor;

        let payload: &[u8] = b"hello world";

        // Compress with lz4 using the block format
        let compressed = lz4_flex::compress(payload);

        // Build a header with corrupted uncompressed_length (1 byte too large)
        let data_length = compressed.len() as u32;
        let uncompressed_length_correct = payload.len() as u32;
        let uncompressed_length_corrupted = uncompressed_length_correct + 1;

        let checksum = Checksum::from_raw(crate::hash::hash128(&compressed));

        let header = Header {
            data_length,
            uncompressed_length: uncompressed_length_corrupted,
            checksum,
            ..Header::test_dummy(BlockType::Data)
        };

        let mut buf = header.encode_into_vec();
        buf.extend_from_slice(&compressed);

        let mut cursor = Cursor::new(buf);
        let result = Block::from_reader(
            &mut cursor,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        match result {
            Err(crate::Error::Decompress(CompressionType::Lz4)) => { /* expected */ }
            Ok(_) => panic!("expected Error::Decompress, but got Ok(Block)"),
            Err(other) => panic!("expected Error::Decompress, got different error: {other:?}"),
        }
    }

    #[test]
    #[cfg(feature = "lz4")]
    fn block_from_file_reject_absurd_uncompressed_length() {
        use crate::coding::Encode;
        use std::io::Write;

        let mut buf = vec![];
        Block::write_into(
            &mut buf,
            b"hello",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        )
        .unwrap();

        // Tamper: set uncompressed_length to u32::MAX.
        // The block checksum only covers the compressed payload bytes; it does not include
        // header fields. The header itself has its own checksum, which we recompute below
        // by re-encoding the modified header, so the tampered block remains internally
        // consistent while exercising the DecompressedSizeTooLarge path.
        let mut reader = &buf[..];
        let mut header = Header::decode_from(&mut reader).unwrap();
        let compressed_payload: Vec<u8> = reader.to_vec();

        header.uncompressed_length = u32::MAX;
        let mut tampered = header.encode_into_vec();
        tampered.extend_from_slice(&compressed_payload);

        let mut tmp = tempfile::NamedTempFile::new().unwrap();
        tmp.write_all(&tampered).unwrap();
        tmp.flush().unwrap();
        let file = std::fs::File::open(tmp.path()).unwrap();

        let handle = crate::table::BlockHandle::new(BlockOffset(0), tampered.len() as u32);
        let result = Block::from_file(
            &file,
            handle,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        assert!(
            matches!(&result, Err(crate::Error::DecompressedSizeTooLarge { .. })),
            "expected DecompressedSizeTooLarge, got: {:?}",
            result.err(),
        );
    }

    #[test]
    #[cfg(feature = "lz4")]
    fn block_from_file_zero_uncompressed_length_with_data_fails_decompress() {
        use crate::coding::Encode;
        use std::io::Write;

        let mut buf = vec![];
        Block::write_into(
            &mut buf,
            b"hello",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        )
        .unwrap();

        let mut reader = &buf[..];
        let mut header = Header::decode_from(&mut reader).unwrap();
        let compressed_payload: Vec<u8> = reader.to_vec();

        header.uncompressed_length = 0;
        let mut tampered = header.encode_into_vec();
        tampered.extend_from_slice(&compressed_payload);

        let mut tmp = tempfile::NamedTempFile::new().unwrap();
        tmp.write_all(&tampered).unwrap();
        tmp.flush().unwrap();
        let file = std::fs::File::open(tmp.path()).unwrap();

        let handle = crate::table::BlockHandle::new(BlockOffset(0), tampered.len() as u32);
        let result = Block::from_file(
            &file,
            handle,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Lz4,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        assert!(
            matches!(&result, Err(crate::Error::Decompress(_))),
            "expected Decompress error, got: {:?}",
            result.err(),
        );
    }

    #[test]
    fn block_from_reader_reject_absurd_data_length() {
        use crate::coding::Encode;

        let mut buf = vec![];
        Block::write_into(
            &mut buf,
            b"hello",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::None,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        )
        .unwrap();

        let mut reader = &buf[..];
        let mut header = Header::decode_from(&mut reader).unwrap();
        let payload: Vec<u8> = reader.to_vec();

        // Set data_length past the limit (no encryption → overhead is 0)
        header.data_length = MAX_DECOMPRESSION_SIZE + 1;
        let mut tampered = header.encode_into_vec();
        tampered.extend_from_slice(&payload);

        let mut r = &tampered[..];
        let result = Block::from_reader(
            &mut r,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::None,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        assert!(
            matches!(&result, Err(crate::Error::DecompressedSizeTooLarge { .. })),
            "expected DecompressedSizeTooLarge, got: {:?}",
            result.err(),
        );
    }

    #[test]
    fn block_from_file_reject_oversized_handle() {
        use std::io::Write;

        let mut tmp = tempfile::NamedTempFile::new().unwrap();
        tmp.write_all(b"dummy").unwrap();
        tmp.flush().unwrap();
        let file = std::fs::File::open(tmp.path()).unwrap();

        let handle = crate::table::BlockHandle::new(BlockOffset(0), u32::MAX);
        let result = Block::from_file(
            &file,
            handle,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::None,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        assert!(
            matches!(&result, Err(crate::Error::DecompressedSizeTooLarge { .. })),
            "expected DecompressedSizeTooLarge, got: {:?}",
            result.err(),
        );
    }

    #[test]
    #[cfg(zstd_any)]
    fn zstd_corrupted_uncompressed_length_triggers_decompress_error() {
        use crate::coding::Encode;
        use std::io::Cursor;

        let payload: &[u8] = b"hello world";

        // Fully-qualified path resolves the trait method unambiguously without
        // needing `use CompressionProvider` in this test module scope.
        let compressed =
            crate::compression::ZstdBackend::compress(payload, 3).expect("zstd compress failed");

        let data_length = compressed.len() as u32;
        let uncompressed_length_corrupted = payload.len() as u32 + 1;

        let checksum = Checksum::from_raw(crate::hash::hash128(&compressed));

        let header = Header {
            data_length,
            uncompressed_length: uncompressed_length_corrupted,
            checksum,
            ..Header::test_dummy(BlockType::Data)
        };

        let mut buf = header.encode_into_vec();
        buf.extend_from_slice(&compressed);

        let mut cursor = Cursor::new(buf);
        let result = Block::from_reader(
            &mut cursor,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Zstd(3),
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        match result {
            Err(crate::Error::Decompress(CompressionType::Zstd(_))) => { /* expected */ }
            Ok(_) => panic!("expected Error::Decompress, but got Ok(Block)"),
            Err(other) => panic!("expected Error::Decompress, got different error: {other:?}"),
        }
    }

    #[test]
    #[cfg(zstd_any)]
    fn zstd_decreased_uncompressed_length_triggers_decompress_error() {
        use crate::coding::Encode;
        use std::io::Cursor;

        let payload: &[u8] = b"hello world hello world hello world";

        let compressed =
            crate::compression::ZstdBackend::compress(payload, 3).expect("zstd compress failed");

        let data_length = compressed.len() as u32;
        // Set uncompressed_length smaller than real decompressed size.
        // The backend decompresses into a buffer of this size; the real output
        // exceeds it, triggering the capacity/length mismatch error.
        let uncompressed_length_too_small = payload.len() as u32 - 1;

        let checksum = Checksum::from_raw(crate::hash::hash128(&compressed));

        let header = Header {
            data_length,
            uncompressed_length: uncompressed_length_too_small,
            checksum,
            ..Header::test_dummy(BlockType::Data)
        };

        let mut buf = header.encode_into_vec();
        buf.extend_from_slice(&compressed);

        let mut cursor = Cursor::new(buf);
        let result = Block::from_reader(
            &mut cursor,
            crate::table::block::BlockIdentity::for_test(0, crate::table::block::BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Zstd(3),
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );

        match result {
            Err(crate::Error::Decompress(CompressionType::Zstd(_))) => { /* expected */ }
            Ok(_) => panic!("expected Error::Decompress, but got Ok(Block)"),
            Err(other) => panic!("expected Error::Decompress, got different error: {other:?}"),
        }
    }

    #[test]
    #[cfg(zstd_any)]
    fn block_roundtrip_zstd() -> crate::Result<()> {
        let mut writer = vec![];

        Block::write_into(
            &mut writer,
            b"abcdefabcdefabcdef",
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Zstd(3),
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;

        {
            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    None,
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(b"abcdefabcdefabcdef", &*block.data);
        }

        Ok(())
    }

    #[test]
    fn block_write_rejects_oversized_payload() {
        let oversized = vec![0u8; MAX_DECOMPRESSION_SIZE as usize + 1];
        let mut sink = std::io::sink();
        let result = Block::write_into(
            &mut sink,
            &oversized,
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::None,
                None,
                #[cfg(zstd_any)]
                None,
            )
            .unwrap(),
        );
        assert!(
            matches!(result, Err(crate::Error::DecompressedSizeTooLarge { .. })),
            "expected DecompressedSizeTooLarge, got: {result:?}",
        );
    }

    #[test]
    #[cfg(zstd_any)]
    fn block_roundtrip_zstd_large_data() -> crate::Result<()> {
        let data = vec![0xABu8; 64 * 1024]; // 64KB
        let mut writer = vec![];

        Block::write_into(
            &mut writer,
            &data,
            crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
            &crate::table::block::BlockTransform::from_parts(
                CompressionType::Zstd(3),
                None,
                #[cfg(zstd_any)]
                None,
            )?,
        )?;

        // Verify compression actually reduced size
        assert!(
            writer.len() < data.len(),
            "zstd should compress repeated data"
        );

        {
            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    None,
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(&*block.data, &data[..]);
        }

        Ok(())
    }

    // --- Encrypted block roundtrip tests ---
    // These exercise the encrypt_vec/decrypt_vec code paths in write_into,
    // from_reader, and from_file that are untouched by the non-encrypted tests.
    //
    // NOTE: The tempfile + write + reopen + handle pattern is duplicated across
    // from_file tests (both encrypted and non-encrypted). Tracked in #127.

    #[cfg(feature = "encryption")]
    mod encrypted {
        use crate::table::block::*;

        fn test_provider() -> crate::encryption::Aes256GcmProvider {
            crate::encryption::Aes256GcmProvider::new(&[0x42; 32])
        }

        #[test]
        fn block_roundtrip_encrypted_uncompressed() -> crate::Result<()> {
            let enc = test_provider();
            let data = b"plaintext block data for encryption test";
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        #[cfg(feature = "lz4")]
        fn block_roundtrip_encrypted_lz4() -> crate::Result<()> {
            let enc = test_provider();
            let data = b"abcdefabcdefabcdef";
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Lz4,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Lz4,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        #[cfg(zstd_any)]
        fn block_roundtrip_encrypted_zstd() -> crate::Result<()> {
            let enc = test_provider();
            let data = b"abcdefabcdefabcdef";
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        fn block_from_file_encrypted_uncompressed() -> crate::Result<()> {
            let enc = test_provider();
            let data = b"plaintext block data for from_file encryption test";
            let tmp = super::write_block_to_tempfile(
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            let block = Block::from_file(
                &tmp.file,
                tmp.handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        #[cfg(feature = "lz4")]
        fn block_from_file_encrypted_lz4() -> crate::Result<()> {
            let enc = test_provider();
            let data = b"abcdefabcdefabcdef";
            let tmp = super::write_block_to_tempfile(
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Lz4,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            let block = Block::from_file(
                &tmp.file,
                tmp.handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Lz4,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        #[cfg(zstd_any)]
        fn block_from_file_encrypted_zstd() -> crate::Result<()> {
            let enc = test_provider();
            let data = b"abcdefabcdefabcdef";
            let tmp = super::write_block_to_tempfile(
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            let block = Block::from_file(
                &tmp.file,
                tmp.handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        fn block_from_file_encrypted_wrong_key_fails() -> crate::Result<()> {
            let enc_write = test_provider();
            let enc_read = crate::encryption::Aes256GcmProvider::new(&[0x99; 32]);
            let data = b"encrypted block data";
            let tmp = super::write_block_to_tempfile(
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc_write),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            let result = Block::from_file(
                &tmp.file,
                tmp.handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc_read),
                    #[cfg(zstd_any)]
                    None,
                )?,
            );
            assert!(
                matches!(result, Err(crate::Error::Decrypt(_))),
                "expected Decrypt error for wrong key, got: {:?}",
                result.err(),
            );
            Ok(())
        }

        #[test]
        fn block_from_reader_encrypted_wrong_key_fails() -> crate::Result<()> {
            let enc_write = test_provider();
            let enc_read = crate::encryption::Aes256GcmProvider::new(&[0x99; 32]);
            let data = b"encrypted block data";
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc_write),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            let mut reader = &writer[..];
            let result = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc_read),
                    #[cfg(zstd_any)]
                    None,
                )?,
            );
            assert!(
                matches!(result, Err(crate::Error::Decrypt(_))),
                "expected Decrypt error for wrong key, got: {:?}",
                result.err(),
            );
            Ok(())
        }

        #[test]
        fn block_from_file_encrypted_checksum_tamper_detected() -> crate::Result<()> {
            use std::io::Write;

            let enc = test_provider();
            let data = b"data for tamper test";
            let mut buf = vec![];
            let header = Block::write_into(
                &mut buf,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            // Tamper a byte in the encrypted payload (after header)
            let mid = Header::MIN_LEN + 1;
            if mid < buf.len() {
                #[expect(clippy::indexing_slicing, reason = "mid < buf.len() checked above")]
                {
                    buf[mid] ^= 0xFF;
                }
            }

            let dir = tempfile::tempdir()?;
            let path = dir.path().join("block");
            let mut file = std::fs::File::create(&path)?;
            file.write_all(&buf)?;
            file.sync_all()?;
            drop(file);

            let file = std::fs::File::open(&path)?;
            let handle = crate::table::BlockHandle::new(BlockOffset(0), header.on_disk_size());
            let result = Block::from_file(
                &file,
                handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            );
            assert!(
                matches!(result, Err(crate::Error::ChecksumMismatch { .. })),
                "expected ChecksumMismatch for tampered data, got: {:?}",
                result.err(),
            );
            Ok(())
        }

        #[test]
        fn block_from_file_encrypted_undersized_handle_rejected() -> crate::Result<()> {
            use std::io::Write;

            let enc = test_provider();
            let dir = tempfile::tempdir()?;
            let path = dir.path().join("block");
            let mut file = std::fs::File::create(&path)?;
            file.write_all(b"tiny")?;
            file.sync_all()?;
            drop(file);

            let file = std::fs::File::open(&path)?;
            // Handle size smaller than Header::MIN_LEN
            let handle = crate::table::BlockHandle::new(BlockOffset(0), 2);
            let result = Block::from_file(
                &file,
                handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            );

            assert!(
                matches!(result, Err(crate::Error::InvalidHeader(_))),
                "expected InvalidHeader for undersized handle, got: {:?}",
                result.err(),
            );
            Ok(())
        }

        #[test]
        fn block_from_file_encrypted_uncompressed_large_payload() -> crate::Result<()> {
            let enc = test_provider();
            let data = vec![0xBB_u8; 32 * 1024]; // 32 KiB
            let tmp = super::write_block_to_tempfile(
                &data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            let block = Block::from_file(
                &tmp.file,
                tmp.handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(&*block.data, &data[..]);
            Ok(())
        }

        #[test]
        fn block_roundtrip_encrypted_uncompressed_large() -> crate::Result<()> {
            let enc = test_provider();
            let data = vec![0xCC_u8; 32 * 1024]; // 32 KiB
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                &data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::None,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(&*block.data, &data[..]);
            Ok(())
        }

        #[test]
        #[cfg(feature = "lz4")]
        fn block_roundtrip_encrypted_lz4_large() -> crate::Result<()> {
            let enc = test_provider();
            let data = vec![0xDD_u8; 32 * 1024]; // 32 KiB
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                &data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Lz4,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Lz4,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(&*block.data, &data[..]);
            Ok(())
        }

        #[test]
        #[cfg(zstd_any)]
        fn block_roundtrip_encrypted_zstd_large() -> crate::Result<()> {
            let enc = test_provider();
            let data = vec![0xEE_u8; 32 * 1024]; // 32 KiB
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                &data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    CompressionType::Zstd(3),
                    Some(&enc),
                    #[cfg(zstd_any)]
                    None,
                )?,
            )?;
            assert_eq!(&*block.data, &data[..]);
            Ok(())
        }
    }

    #[cfg(feature = "zstd")]
    mod zstd_dict {
        use super::*;
        use crate::compression::ZstdDictionary;
        use test_log::test;

        fn test_dict() -> ZstdDictionary {
            let mut samples = Vec::new();
            for i in 0u32..500 {
                samples.extend_from_slice(format!("key-{i:05}val-{i:05}").as_bytes());
            }
            ZstdDictionary::new(&samples)
        }

        fn test_compression(dict: &ZstdDictionary) -> CompressionType {
            CompressionType::ZstdDict {
                level: 3,
                dict_id: dict.id(),
            }
        }

        #[test]
        fn block_roundtrip_zstd_dict_reader() -> crate::Result<()> {
            let dict = test_dict();
            let compression = test_compression(&dict);
            let data = b"abcdefabcdefabcdef";
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    None,
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    None,
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        fn block_roundtrip_zstd_dict_file() -> crate::Result<()> {
            use std::io::Write;

            let dict = test_dict();
            let compression = test_compression(&dict);
            let data = b"abcdefabcdefabcdef";
            let mut buf = vec![];
            let header = Block::write_into(
                &mut buf,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    None,
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;

            let dir = tempfile::tempdir()?;
            let path = dir.path().join("block");
            let mut file = std::fs::File::create(&path)?;
            file.write_all(&buf)?;
            file.sync_all()?;
            drop(file);

            let file = std::fs::File::open(&path)?;
            let handle = crate::table::BlockHandle::new(BlockOffset(0), header.on_disk_size());
            let block = Block::from_file(
                &file,
                handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    None,
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        fn block_roundtrip_zstd_dict_large_data() -> crate::Result<()> {
            let dict = test_dict();
            let compression = test_compression(&dict);
            let data = vec![0xAB_u8; 64 * 1024]; // 64 KiB
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                &data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    None,
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;

            assert!(
                writer.len() < data.len(),
                "dict compression should reduce size"
            );

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    None,
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;
            assert_eq!(&*block.data, &data[..]);
            Ok(())
        }

        #[test]
        fn block_zstd_dict_wrong_dict_returns_error() {
            // Companion test to
            // `block_transform_from_parts_zstd_dict_missing_returns_error`
            // (below): both assert the BlockTransform::from_parts
            // check that used to live inside Block::write_into /
            // from_reader for the ZstdDict codec. The dict-missing
            // case exercises the `None` half of the dict argument;
            // this one exercises the cross-check between the
            // supplied dictionary id and the
            // ZstdDict { dict_id } discriminator. Both assert
            // directly on the transform-construction result; no
            // Block I/O call is needed to exercise the mismatch
            // path.
            let dict = test_dict();
            let compression = test_compression(&dict);
            let wrong_dict = ZstdDictionary::new(b"completely different dictionary bytes");

            let result = crate::table::block::BlockTransform::from_parts(
                compression,
                None,
                Some(&wrong_dict),
            );
            assert!(
                matches!(
                    result,
                    Err(crate::Error::ZstdDictMismatch { got: Some(_), .. })
                ),
                "expected ZstdDictMismatch with got=Some",
            );
        }

        #[test]
        fn block_transform_from_parts_zstd_dict_missing_returns_error() {
            // The runtime dict-presence check that used to live inside
            // Block::write_into / from_reader for the ZstdDict codec
            // is now centralised in BlockTransform::from_parts. The
            // error therefore surfaces at transform-construction time
            // instead of at the Block I/O call; this test verifies
            // that earlier surface — it no longer exercises
            // Block::write_into / from_reader at all, hence the test
            // name describes from_parts rather than block_write_*.
            // (See block_zstd_dict_wrong_dict_returns_error above for
            // the matching "wrong dict id" path.)
            let dict = test_dict();
            let compression = test_compression(&dict);

            // Try to construct the read-side transform without
            // providing the dict the codec needs.
            let result = crate::table::block::BlockTransform::from_parts(compression, None, None);
            // BlockTransform holds `&dyn EncryptionProvider` which
            // doesn't impl Debug, so we can't print the whole result;
            // surface just the Err side (which IS Debug) on mismatch.
            // Match on `&result` + `.as_ref().err()` so the variant
            // check and the formatter borrow the same value — no
            // need to reason about whether the matches! patterns
            // bind by value.
            assert!(
                matches!(
                    &result,
                    Err(crate::Error::ZstdDictMismatch { got: None, .. })
                ),
                "expected ZstdDictMismatch, got: {:?}",
                result.as_ref().err(),
            );
        }

        #[test]
        #[cfg(feature = "encryption")]
        fn block_roundtrip_zstd_dict_encrypted_reader() -> crate::Result<()> {
            let enc = crate::Aes256GcmProvider::new(&[0x42; 32]);
            let dict = test_dict();
            let compression = test_compression(&dict);
            let data = b"encrypted-dict-compressed-data-for-test";
            let mut writer = vec![];

            Block::write_into(
                &mut writer,
                data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;
            assert_eq!(data, &*block.data);
            Ok(())
        }

        #[test]
        #[cfg(feature = "encryption")]
        fn block_roundtrip_zstd_dict_encrypted_file() -> crate::Result<()> {
            use std::io::Write;

            let enc = crate::Aes256GcmProvider::new(&[0x42; 32]);
            let dict = test_dict();
            let compression = test_compression(&dict);
            let data = vec![0xCC_u8; 16 * 1024]; // 16 KiB
            let mut buf = vec![];
            let header = Block::write_into(
                &mut buf,
                &data,
                crate::table::block::BlockIdentity::for_test(0, BlockType::Data),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;

            let dir = tempfile::tempdir()?;
            let path = dir.path().join("block");
            let mut file = std::fs::File::create(&path)?;
            file.write_all(&buf)?;
            file.sync_all()?;
            drop(file);

            let file = std::fs::File::open(&path)?;
            let handle = crate::table::BlockHandle::new(BlockOffset(0), header.on_disk_size());
            let block = Block::from_file(
                &file,
                handle,
                crate::table::block::BlockIdentity::for_test(
                    0,
                    crate::table::block::BlockType::Data,
                ),
                &crate::table::block::BlockTransform::from_parts(
                    compression,
                    Some(&enc),
                    #[cfg(zstd_any)]
                    Some(&dict),
                )?,
            )?;
            assert_eq!(&*block.data, &data[..]);
            Ok(())
        }
    }

    /// Page ECC integration tests — write a block with the
    /// `BlockTransform::*Ecc` variant, verify the on-disk layout
    /// round-trips through `Block::from_reader`, and verify that
    /// Reed-Solomon recovery kicks in when the payload bytes are
    /// corrupted between write and read.
    #[cfg(feature = "page_ecc")]
    mod page_ecc {
        use super::*;
        use test_log::test;

        const PAYLOAD: &[u8] = b"the quick brown fox jumps over the lazy dog \
                                 0123456789 the quick brown fox jumps over \
                                 the lazy dog 0123456789";

        #[test]
        fn block_roundtrip_plain_ecc_clean_read() -> crate::Result<()> {
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;

            assert!(
                header.block_flags & crate::table::block::header::block_flags::ECC_PARITY != 0,
                "PlainEcc writer must set the ECC_PARITY flag",
            );
            assert_eq!(
                writer.len(),
                header.on_disk_size() as usize,
                "on-disk size must equal header + payload + derived parity length",
            );

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;
            assert_eq!(&*block.data, PAYLOAD);
            Ok(())
        }

        #[test]
        fn block_roundtrip_plain_ecc_recovers_from_single_byte_flip() -> crate::Result<()> {
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;

            // Flip a single byte inside the payload region (after
            // the header, before the parity trailer) so the on-disk
            // bytes' XXH3 no longer matches header.checksum but the
            // recoverable shape (1 of 6 shards corrupted) holds.
            let header_len = Header::MIN_LEN;
            let flip_at = header_len + (header.data_length as usize) / 2;
            writer[flip_at] ^= 0xFF;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;
            // ECC recovery reconstructs the original payload despite
            // the in-flight bit-flip.
            assert_eq!(
                &*block.data, PAYLOAD,
                "Reed-Solomon recovery must reconstruct the original \
                 payload from a single-byte data-shard flip",
            );
            Ok(())
        }

        #[test]
        fn block_roundtrip_secded_clean_read() -> crate::Result<()> {
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::SECDED),
            )?;

            assert!(
                header.block_flags & crate::table::block::header::block_flags::ECC_PARITY != 0,
                "SECDED writer must set the ECC_PARITY flag",
            );
            // One parity byte per 8-byte word: header + payload + ceil(N/8).
            // `on_disk_size()` hardcodes the RS(4,2) default; the scheme-aware
            // `on_disk_size_with` sizes the SECDED trailer.
            assert_eq!(
                writer.len(),
                header.on_disk_size_with(Some(EccParams::SECDED)) as usize,
                "on-disk size must equal header + payload + SECDED parity length",
            );

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::SECDED),
            )?;
            assert_eq!(&*block.data, PAYLOAD);
            Ok(())
        }

        #[test]
        fn block_roundtrip_secded_recovers_from_single_bit_flip() -> crate::Result<()> {
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::SECDED),
            )?;

            // Flip a SINGLE bit inside the payload region: SECDED heals one bit
            // per 8-byte word, so a single-bit flip is recoverable (a whole-byte
            // flip would be 8 bit-errors in one word — uncorrectable).
            let flip_at = Header::MIN_LEN + (header.data_length as usize) / 2;
            writer[flip_at] ^= 0x01;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::SECDED),
            )?;
            assert_eq!(
                &*block.data, PAYLOAD,
                "SECDED must heal a single-bit payload flip",
            );
            Ok(())
        }

        #[test]
        fn block_roundtrip_secded_unrecoverable_on_double_bit_flip() -> crate::Result<()> {
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::SECDED),
            )?;

            // Two bit errors in the same byte (hence the same 8-byte word):
            // SECDED detects but cannot correct — the read must fail rather
            // than return miscorrected data.
            let flip_at = Header::MIN_LEN + (header.data_length as usize) / 2;
            writer[flip_at] ^= 0x03;

            let mut reader = &writer[..];
            let result = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::SECDED),
            );
            assert!(
                matches!(&result, Err(crate::Error::PageEccUnrecoverable { .. })),
                "a double-bit error in one word must be detected as unrecoverable \
                 (got ok={})",
                result.is_ok(),
            );
            Ok(())
        }

        #[test]
        fn block_from_file_plain_ecc_recovers_from_single_byte_flip() -> crate::Result<()> {
            let tmp = super::write_block_to_tempfile(
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;
            let path = tmp.dir.path().join("block");

            // Flip one byte inside the payload region (after the
            // header, before the parity trailer).
            let mut bytes = std::fs::read(&path)?;
            let payload_start = Header::MIN_LEN;
            bytes[payload_start + 3] ^= 0x80;
            std::fs::write(&path, &bytes)?;

            // Re-open and read via from_file. ECC recovery should
            // reconstruct the original payload.
            let file = std::fs::File::open(&path)?;
            let block = Block::from_file(
                &file,
                tmp.handle,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;
            assert_eq!(&*block.data, PAYLOAD);
            Ok(())
        }

        /// `from_file_with_status` reports `EccStatus::Corrected` when a read
        /// repaired the block via ECC, and `EccStatus::Ok` for a clean read.
        /// This is the auto-heal (#411) signal: a healed read returns the
        /// correct bytes AND flags that the on-disk copy still holds the fault.
        #[test]
        fn from_file_with_status_reports_corrected_after_ecc_repair() -> crate::Result<()> {
            let transform = BlockTransform::PlainEcc(EccParams::RS_4_2);
            let tmp = super::write_block_to_tempfile(
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &transform,
            )?;
            let path = tmp.dir.path().join("block");

            // Clean read first: no repair → EccStatus::Ok.
            {
                let file = std::fs::File::open(&path)?;
                let (block, status) = Block::from_file_with_status(
                    &file,
                    tmp.handle,
                    BlockIdentity::for_test(0, BlockType::Data),
                    &transform,
                )?;
                assert_eq!(&*block.data, PAYLOAD);
                assert_eq!(status, EccStatus::Ok, "clean read must not flag a repair");
            }

            // Flip one payload byte so the read must repair via RS parity.
            let mut bytes = std::fs::read(&path)?;
            bytes[Header::MIN_LEN + 3] ^= 0x80;
            std::fs::write(&path, &bytes)?;

            let file = std::fs::File::open(&path)?;
            let (block, status, recovery) = Block::from_file_with_recovery(
                &file,
                tmp.handle,
                BlockIdentity::for_test(0, BlockType::Data),
                &transform,
            )?;
            assert_eq!(&*block.data, PAYLOAD, "repaired bytes must equal original");
            assert_eq!(
                status,
                EccStatus::Corrected,
                "a repaired read reports Corrected"
            );
            assert_eq!(
                recovery,
                Some(EccRecoveryKind::Shard),
                "an RS repair is attributed to the shard mechanism",
            );
            Ok(())
        }

        /// A single-bit flip in a SEC-DED-protected block is healed by the
        /// SEC-DED fast path and reported as `Corrected(Secded)`, distinct from
        /// the RS shard path. This is the kind-attribution the unified recovery
        /// metric relies on (the per-kind counter is bumped from this status at
        /// the `load_block` / scrub / partial-decode call sites).
        #[test]
        fn from_file_with_status_reports_secded_kind_after_single_bit_heal() -> crate::Result<()> {
            let transform = BlockTransform::PlainEcc(EccParams::SECDED);
            let tmp = super::write_block_to_tempfile(
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &transform,
            )?;
            let path = tmp.dir.path().join("block");

            // Flip a SINGLE bit so the SEC-DED per-word code heals it.
            let mut bytes = std::fs::read(&path)?;
            bytes[Header::MIN_LEN + 3] ^= 0x01;
            std::fs::write(&path, &bytes)?;

            let file = std::fs::File::open(&path)?;
            let (block, status, recovery) = Block::from_file_with_recovery(
                &file,
                tmp.handle,
                BlockIdentity::for_test(0, BlockType::Data),
                &transform,
            )?;
            assert_eq!(
                &*block.data, PAYLOAD,
                "SEC-DED must heal the single-bit flip"
            );
            assert_eq!(
                status,
                EccStatus::Corrected,
                "a repaired read reports Corrected"
            );
            assert_eq!(
                recovery,
                Some(EccRecoveryKind::Secded),
                "a SEC-DED single-bit heal is attributed to the SEC-DED mechanism",
            );
            Ok(())
        }

        /// Three-state read contract: a block written WITH a parity trailer,
        /// read back by a reader that does NOT recognize the scheme (a `Plain`
        /// transform — as happens when the per-SST descriptor decodes to an
        /// unsupported scheme), must still return the payload (framed by
        /// `data_length`, checksum-verified) and report
        /// `EccStatus::Unrecognized` — a soft warning, not a read error. Read
        /// with the matching scheme, it reports `EccStatus::Ok`.
        #[test]
        fn from_file_with_status_soft_warns_on_unrecognized_trailer() -> crate::Result<()> {
            // A non-default scheme on purpose: the read path is descriptor-
            // driven, never an implicit RS(4,2).
            let scheme = EccParams::try_new(8, 2).expect("valid shards");
            let tmp = super::write_block_to_tempfile(
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(scheme),
            )?;

            // Reader recognizes the scheme → clean read.
            let (block, status) = Block::from_file_with_status(
                &tmp.file,
                tmp.handle,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(scheme),
            )?;
            assert_eq!(&*block.data, PAYLOAD);
            assert_eq!(status, EccStatus::Ok);

            // Reader does NOT recognize the trailer (Plain transform): the
            // opaque trailer is excluded from the payload, the checksum still
            // verifies, and the status is a warning rather than an error.
            let (block, status) = Block::from_file_with_status(
                &tmp.file,
                tmp.handle,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PLAIN,
            )?;
            assert_eq!(
                &*block.data, PAYLOAD,
                "payload reads despite unknown trailer"
            );
            assert_eq!(status, EccStatus::Unrecognized);
            Ok(())
        }

        /// Regression: a RECOGNIZED ECC layout must require an exact trailer
        /// length — including the empty-payload case where the parity length is
        /// zero. Extra trailing bytes on such a block are corruption and must
        /// FAIL, not be softened to `EccStatus::Unrecognized` (which keying the
        /// decision off `ecc_length == 0` instead of "is the layout recognized"
        /// would wrongly do).
        #[test]
        fn from_file_recognized_empty_block_rejects_extra_trailer() -> crate::Result<()> {
            // Empty payload under a (non-default) recognized scheme →
            // data_length 0, parity length 0, so the on-disk block is just the
            // header (no trailer).
            let scheme = EccParams::try_new(8, 2).expect("valid shards");
            let tmp = super::write_block_to_tempfile(
                b"",
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(scheme),
            )?;
            let path = tmp.dir.path().join("block");
            let base = tmp.handle.size();

            // Append junk so the handle covers a 4-byte trailer that the
            // recognized (zero-parity) layout does not expect.
            let mut bytes = std::fs::read(&path)?;
            bytes.extend_from_slice(&[0xAB, 0xCD, 0xEF, 0x01]);
            std::fs::write(&path, &bytes)?;

            let file = std::fs::File::open(&path)?;
            let handle = crate::table::BlockHandle::new(crate::table::BlockOffset(0), base + 4);
            // `.err()` drops the `Ok((Block, _))` payload (Block is not Debug)
            // so the assert can format the error variant.
            let err = Block::from_file_with_status(
                &file,
                handle,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(scheme),
            )
            .err();
            assert!(
                matches!(err, Some(crate::Error::InvalidHeader("Block"))),
                "recognized zero-parity layout + extra trailer must fail, got {err:?}",
            );
            Ok(())
        }

        /// Same recovery story as `PlainEcc` but with lz4 compression
        /// stacked on top: parity is computed over the
        /// post-compression payload, recovery happens BEFORE
        /// decompression. Catches a regression where the read path
        /// would try to decompress corrupt bytes (lz4 would fail
        /// before recovery had a chance to fire).
        #[cfg(feature = "lz4")]
        #[test]
        fn block_roundtrip_compressed_ecc_recovers_from_byte_flip() -> crate::Result<()> {
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::CompressedEcc(
                    CompressionContext::new(CompressionType::Lz4)?,
                    EccParams::RS_4_2,
                ),
            )?;
            assert!(header.block_flags & crate::table::block::header::block_flags::ECC_PARITY != 0);

            // Flip a byte in the compressed-payload region.
            let header_len = Header::MIN_LEN;
            let flip_at = header_len + (header.data_length as usize) / 2;
            writer[flip_at] ^= 0x55;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::CompressedEcc(
                    CompressionContext::new(CompressionType::Lz4)?,
                    EccParams::RS_4_2,
                ),
            )?;
            assert_eq!(
                &*block.data, PAYLOAD,
                "ECC must recover the compressed bytes BEFORE lz4 \
                 decompression, otherwise lz4 would fail on corrupt input",
            );
            Ok(())
        }

        /// ECC-protected encrypted roundtrip with a single-byte
        /// ciphertext flip. Parity is computed over the ciphertext,
        /// so recovery must produce byte-exact reconstruction —
        /// AEAD authentication fails on even a one-bit mismatch.
        /// This catches a regression where ECC recovery rebuilds an
        /// arithmetically valid Reed-Solomon shard that doesn't
        /// bit-identically reproduce the original ciphertext.
        #[cfg(feature = "encryption")]
        #[test]
        fn block_roundtrip_encrypted_ecc_recovers_from_byte_flip() -> crate::Result<()> {
            let enc = crate::encryption::Aes256GcmProvider::new(&[0x42; 32]);
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::EncryptedEcc(&enc, EccParams::RS_4_2),
            )?;
            assert!(header.block_flags & crate::table::block::header::block_flags::ECC_PARITY != 0);

            // Flip one byte in the ciphertext region.
            let header_len = Header::MIN_LEN;
            let flip_at = header_len + (header.data_length as usize) / 2;
            writer[flip_at] ^= 0x21;

            let mut reader = &writer[..];
            let block = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::EncryptedEcc(&enc, EccParams::RS_4_2),
            )?;
            assert_eq!(
                &*block.data, PAYLOAD,
                "ECC must reconstruct ciphertext byte-exactly so AEAD \
                 authentication succeeds on the recovered bytes",
            );
            Ok(())
        }

        /// Asserts the unrecoverable path surfaces
        /// `Error::PageEccUnrecoverable`. Corrupts enough on-disk
        /// bytes to take out more than the RS(4, 2) scheme can
        /// recover (≥ 3 data shards damaged), so every C(6, 4)
        /// subset trial decode fails the xxh3 oracle and
        /// `try_recover` exhausts all 15 candidates.
        #[test]
        fn block_roundtrip_plain_ecc_unrecoverable_when_too_many_shards_corrupt()
        -> crate::Result<()> {
            let mut writer = vec![];
            let header = Block::write_into(
                &mut writer,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;

            // Shard size in bytes — same formula as crate::ecc::shard_bytes
            // (ceil(payload_len / 4) rounded up to even).
            let payload_len = header.data_length as usize;
            let shard_bytes = ((payload_len.div_ceil(4)) + 1) & !1;

            // Flip one byte in EACH of the first 3 data shards.
            // RS(4, 2) recovers up to 2 missing shards; with 3
            // corrupted data shards no subset of 4 intact shards
            // reconstructs the original.
            let payload_start = Header::MIN_LEN;
            for shard_idx in 0..3 {
                let pos = payload_start + shard_idx * shard_bytes;
                if pos < writer.len() {
                    writer[pos] ^= 0xFF;
                }
            }

            let mut reader = &writer[..];
            let result = Block::from_reader(
                &mut reader,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            );
            match result {
                Ok(_) => panic!(
                    "3-shard corruption must exceed RS(4,2) recovery capacity, \
                     but from_reader returned Ok"
                ),
                Err(crate::Error::PageEccUnrecoverable { .. }) => {}
                Err(e) => panic!("expected PageEccUnrecoverable, got {e:?}"),
            }
            Ok(())
        }

        #[test]
        fn ecc_parity_bit_agrees_with_emitted_parity_length() -> crate::Result<()> {
            use crate::table::block::header::block_flags;

            // Empty payload: the Reed-Solomon encoder short-circuits to a
            // zero-length parity trailer even under PlainEcc, so the
            // presence-authoritative ECC_PARITY bit must stay CLEAR — and the
            // derived on-disk size carries no parity (header + 0 payload).
            let mut empty_buf = vec![];
            let empty = Block::write_into(
                &mut empty_buf,
                &[],
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;
            assert_eq!(
                empty.block_flags & block_flags::ECC_PARITY,
                0,
                "ECC_PARITY must be clear when no parity trailer is emitted",
            );
            assert_eq!(
                empty_buf.len(),
                empty.on_disk_size() as usize,
                "on-disk size matches the derived (zero) parity length",
            );
            assert_eq!(
                empty.on_disk_size() as usize,
                Header::MIN_LEN,
                "empty payload emits no parity, so on-disk size is just the header",
            );

            // Non-empty payload: parity is emitted, so the bit is set and the
            // derived on-disk size includes the parity trailer.
            let mut full_buf = vec![];
            let full = Block::write_into(
                &mut full_buf,
                PAYLOAD,
                BlockIdentity::for_test(0, BlockType::Data),
                &BlockTransform::PlainEcc(EccParams::RS_4_2),
            )?;
            assert_ne!(
                full.block_flags & block_flags::ECC_PARITY,
                0,
                "ECC_PARITY must be set when a parity trailer is emitted",
            );
            assert_eq!(
                full_buf.len(),
                full.on_disk_size() as usize,
                "on-disk size matches header + payload + derived parity",
            );
            assert!(
                full.on_disk_size() as usize > Header::MIN_LEN + full.data_length as usize,
                "non-empty payload emits a parity trailer beyond header + payload",
            );
            Ok(())
        }
    }
}