musefs_core/

reader.rs

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::Mutex;

use musefs_db::convert::usize_from;
use musefs_db::{Db, Format};
use musefs_format::flac::{self, MetadataBlock};
use musefs_format::{BinaryTagInput, RegionLayout, Segment, mp3, mp4, wav};
use quick_cache::Weighter;
use quick_cache::sync::Cache;

use crate::error::{CoreError, Result};
use crate::facade::Mode;
use crate::freshness::BackingStamp;
use crate::mapping::{tags_to_inputs, track_art_to_inputs};
use crate::ogg_index::serve_ogg_window;

/// A fully resolved synthesized file: its segment layout, total size, the
/// content version it was built from, and where the backing audio lives.
#[derive(Debug)]
pub struct ResolvedFile {
    pub layout: RegionLayout,
    pub total_len: u64,
    /// Track id this entry resolves. Lets the stateless read path recheck the
    /// live `content_version` under its WAL snapshot (#502), mirroring the
    /// handle fast path's use of the handle's `track_id`.
    pub track_id: i64,
    pub content_version: i64,
    pub backing_path: PathBuf,
    pub stamp: BackingStamp,
    pub mtime_secs: i64,
    /// One-entry memo of the last patched Ogg page, so consecutive reads skip
    /// re-patching the page straddling a chunk boundary. Empty for non-Ogg files
    /// and reset whenever this resolved entry is rebuilt. (Concrete type spelled
    /// out rather than `ogg_index::LastPageMemo` because that module is private.)
    pub last_page: Mutex<Option<(u64, u64, Vec<u8>)>>,
    /// Approximate resident bytes this entry costs the cache (sum of `Inline`
    /// segment bytes; backing/art/ogg-audio bytes are not resident).
    pub cache_bytes: u64,
    /// Precomputed from the layout: true if any segment is streamed from the DB
    /// by a rowid (`BinaryTag`/`ArtImage`/`OggArtSlice`). Gates the transactional
    /// `content_version` guard in the read fast path so plain Inline/BackingAudio
    /// layouts pay no per-read cost (#502).
    pub streams_db_rowid: bool,
}

/// Weighs an entry by its resident inline bytes. The `.max(1)` is load-bearing:
/// quick_cache ignores zero-weight entries when evicting, and every
/// StructureOnly layout has `cache_bytes == 0`, so an unweighted entry would
/// escape the byte budget entirely.
#[derive(Clone)]
struct CacheBytesWeighter;

impl Weighter<i64, Arc<ResolvedFile>> for CacheBytesWeighter {
    fn weight(&self, _key: &i64, val: &Arc<ResolvedFile>) -> u64 {
        val.cache_bytes.max(1)
    }
}

/// A per-mount cache of resolved files keyed by track id; an entry
/// self-invalidates when the track's `content_version` changes. Backed by
/// quick_cache: S3-FIFO eviction, byte-weighted, internally sharded.
pub struct HeaderCache {
    cache: Cache<i64, Arc<ResolvedFile>, CacheBytesWeighter>,
    mode: Mode,
}

/// Default resident-bytes budget for the header cache (64 MiB).
pub const DEFAULT_CACHE_BUDGET: u64 = 64 * 1024 * 1024;

/// Item-count sizing hint for quick_cache's internal structures (not a bound):
/// the default budget over 4 KiB, a typical inline tag region. The hint has no
/// observable public-API behavior, so its arithmetic carries an equivalent-mutant
/// exclusion in .cargo/mutants.toml (cargo-mutants does mutate const initializers).
const CACHE_ESTIMATED_ITEMS: usize = (DEFAULT_CACHE_BUDGET / 4096) as usize;

fn read_front(path: &Path, n: u64) -> crate::Result<Vec<u8>> {
    use std::io::Read;
    // Fail closed before any allocation/open: a hostile DB row can request an
    // arbitrary `audio_offset`, but no legitimately-scanned file has a front
    // larger than the scanner's probe ceiling. Bounding `n` here also retires a
    // 32-bit `usize_from` truncation footgun.
    if n > crate::scan::MAX_PROBE_BYTES {
        return Err(CoreError::HeaderTooLarge {
            requested: n,
            cap: crate::scan::MAX_PROBE_BYTES,
        });
    }
    crate::metrics::on_open();
    let mut f = std::fs::File::open(path).map_err(|e| CoreError::backing_io(path, e))?;
    let mut buf = vec![0u8; usize_from(n)];
    f.read_exact(&mut buf)
        .map_err(|e| CoreError::backing_io(path, e))?;
    Ok(buf)
}

impl HeaderCache {
    pub fn new(mode: Mode) -> HeaderCache {
        HeaderCache::with_budget(mode, DEFAULT_CACHE_BUDGET)
    }
    pub fn with_budget(mode: Mode, budget: u64) -> HeaderCache {
        HeaderCache {
            cache: Cache::with_weighter(CACHE_ESTIMATED_ITEMS, budget, CacheBytesWeighter),
            mode,
        }
    }
    /// Drop cached resolutions for tracks no longer present (`live` = current ids).
    pub fn retain(&self, live: &HashSet<i64>) {
        self.cache.retain(|id, _| live.contains(id));
    }
    /// Drop one track's cached resolution (changelog-refresh removal path).
    pub fn remove(&self, id: i64) {
        self.cache.remove(&id);
    }
    /// Resolve a track to its layout, caching on a content-version miss. Validation
    /// (`stat`) and synthesis run outside the cache; quick_cache's internal locks
    /// are only touched by the brief get and insert.
    pub fn resolve<M>(&self, db: &Db<M>, track_id: i64) -> Result<Arc<ResolvedFile>> {
        let track = db
            .get_track(track_id)?
            .ok_or(CoreError::TrackNotFound(track_id))?;

        // Always validate the backing file first — a stale file is an error even
        // on a cache hit, because the audio region may have shifted.
        crate::metrics::on_stat();
        let meta = std::fs::metadata(&track.backing_path)
            .map_err(|e| CoreError::backing_io(&track.backing_path, e))?;
        if BackingStamp::from_metadata(&meta) != BackingStamp::from_track(&track) {
            return Err(CoreError::BackingChanged(track.backing_path.clone()));
        }

        if let Some(hit) = self.cache.get(&track_id)
            && hit.content_version == track.content_version
        {
            return Ok(hit);
        }
        let resolved = self.build(db, &track, &meta)?;
        self.cache.insert(track_id, resolved.clone());
        Ok(resolved)
    }
    /// Build a `ResolvedFile` for `track` (synthesis or passthrough). No lock held.
    fn build<M>(
        &self,
        db: &Db<M>,
        track: &musefs_db::Track,
        meta: &std::fs::Metadata,
    ) -> Result<Arc<ResolvedFile>> {
        let (layout, total_len, mtime_secs_val) = match self.mode {
            Mode::StructureOnly => {
                // Pure passthrough: the synthesized "file" is the backing file itself.
                // The stored audio bounds are irrelevant here — the whole file is served
                // verbatim — so they are not validated in this mode.
                let layout = RegionLayout::validated(vec![Segment::BackingAudio {
                    offset: 0,
                    len: meta.len(),
                }])
                .map_err(musefs_format::FormatError::InvalidLayout)?;
                (
                    layout,
                    meta.len(),
                    BackingStamp::from_track(track).display_secs(),
                )
            }
            Mode::Synthesis => {
                // Guard the stored audio bounds before any cast/allocation: a negative
                // bound, or an audio region that runs past the end of the backing file,
                // means the row no longer matches the file. Only synthesis splices at
                // these bounds, so the check is scoped to this mode.
                if track
                    .bounds
                    .audio_offset()
                    .saturating_add(track.bounds.audio_length())
                    > meta.len()
                {
                    return Err(CoreError::BackingChanged(track.backing_path.clone()));
                }

                let inputs = tags_to_inputs(db.get_tags(track.id)?);
                let art_inputs = track_art_to_inputs(db, track.id)?;
                let binary_tag_inputs = crate::mapping::binary_tags_to_inputs(db, track.id)?;

                // FLAC re-reads the front for its preserved structural blocks; MP3 needs no
                // front read — its ID3v2 tag is regenerated entirely from the DB and the
                // Xing/LAME info frame travels with the backing audio.
                let layout = match track.format {
                    Format::Flac => {
                        let rows = db.get_structural_blocks(track.id)?;
                        // Fast path: the structural store holds STREAMINFO/SEEKTABLE and
                        // APPLICATION/CUESHEET stream from value_blob rows. Legacy
                        // fallback (no structural rows yet): carry every preserved block
                        // — including APPLICATION/CUESHEET — inline from the front
                        // re-read, and suppress the streamed binary tags so those blocks
                        // are not emitted twice.
                        let (structural, binary_tags): (Vec<MetadataBlock>, &[BinaryTagInput]) =
                            if rows.is_empty() {
                                let front = read_front(
                                    Path::new(&track.backing_path),
                                    track.bounds.audio_offset(),
                                )?;
                                (flac::read_metadata(&front)?.preserved, &[])
                            } else {
                                let structural = rows
                                    .into_iter()
                                    .filter_map(|b| {
                                        flac::structural_block_type(&b.kind).map(|block_type| {
                                            MetadataBlock {
                                                block_type,
                                                body: b.body,
                                            }
                                        })
                                    })
                                    .collect();
                                (structural, &binary_tag_inputs)
                            };
                        for key in invalid_vorbis_keys(&inputs) {
                            log::warn!(
                                "track {}: dropping tag key {key:?} from Vorbis \
                                 synthesis (not a valid field name)",
                                track.id
                            );
                        }
                        flac::synthesize_layout(
                            &structural,
                            track.bounds.audio_offset(),
                            track.bounds.audio_length(),
                            &inputs,
                            binary_tags,
                            &art_inputs,
                        )?
                    }
                    Format::Mp3 => mp3::synthesize_layout(
                        track.bounds.audio_offset(),
                        track.bounds.audio_length(),
                        &inputs,
                        &binary_tag_inputs,
                        &art_inputs,
                    )?,
                    Format::M4a => {
                        // Read only the structural boxes (ftyp/moov/mdat header) by
                        // seeking — never the (potentially hundreds-of-MB) mdat payload,
                        // which is served from the backing file at read time. The `moov`
                        // box may sit at EOF; the streaming reader skips the mdat payload
                        // to reach it. The resulting layout's leading inline `head` ends
                        // in a deliberately truncated `mdat` header whose payload is the
                        // backing-audio tail.
                        let mut f = std::fs::File::open(&track.backing_path)
                            .map_err(|e| CoreError::backing_io(&track.backing_path, e))?;
                        // `meta` was validated against the tracked size/mtime above,
                        // so reuse it rather than issuing a second fstat.
                        let len = meta.len();
                        let scan = mp4::read_structure_from(&mut f, len).map_err(|e| match e {
                            mp4::Mp4ScanError::Io(io) => {
                                CoreError::backing_io(&track.backing_path, io)
                            }
                            mp4::Mp4ScanError::Format(fe) => CoreError::Format(fe),
                            // Unreachable in practice (an ingested file already
                            // passed the cap at scan, and backing-file drift is
                            // caught by the size/mtime BackingChanged guard first),
                            // but preserve the box/size/cap diagnostics rather than
                            // erasing them into a generic Malformed.
                            mp4::Mp4ScanError::MetadataTooLarge {
                                box_kind,
                                size,
                                cap,
                            } => CoreError::Mp4MetadataTooLarge {
                                box_kind,
                                size,
                                cap,
                            },
                        })?;
                        mp4::synthesize_layout(&scan, &inputs, &binary_tag_inputs, &art_inputs)?
                    }
                    Format::Wav => {
                        // Read only the front (RIFF header + fmt/fact); the data
                        // payload is served from the backing file at read time.
                        let front = read_front(
                            Path::new(&track.backing_path),
                            track.bounds.audio_offset(),
                        )?;
                        let scan = wav::read_structure(&front)?;
                        wav::synthesize_layout(
                            &scan,
                            track.bounds.audio_offset(),
                            track.bounds.audio_length(),
                            &inputs,
                            &binary_tag_inputs,
                            &art_inputs,
                        )?
                    }
                    Format::Opus | Format::Vorbis | Format::OggFlac => {
                        let front = read_front(
                            Path::new(&track.backing_path),
                            track.bounds.audio_offset(),
                        )?;
                        let header = musefs_format::ogg::read_metadata(&front)?;
                        let arts: Vec<musefs_format::ogg::OggArt> = art_inputs
                            .iter()
                            .map(|meta| musefs_format::ogg::OggArt { meta })
                            .collect();
                        let src = crate::mapping::DbArtSource(db);
                        for key in invalid_vorbis_keys(&inputs) {
                            log::warn!(
                                "track {}: dropping tag key {key:?} from Vorbis \
                                 synthesis (not a valid field name)",
                                track.id
                            );
                        }
                        musefs_format::ogg::synthesize_layout(
                            &header,
                            track.bounds.audio_offset(),
                            track.bounds.audio_length(),
                            &inputs,
                            &arts,
                            &src,
                        )?
                    }
                };
                let total = layout.total_len();
                (
                    layout,
                    total,
                    BackingStamp::from_track(track)
                        .display_secs()
                        .max(track.updated_at),
                )
            }
        };

        // Defensive belt-and-suspenders: production layouts are already built via
        // RegionLayout::validated, but re-validate at the cache boundary so a future
        // construction path that skips validation cannot poison the cache.
        layout
            .validate()
            .map_err(musefs_format::FormatError::InvalidLayout)?;

        let cache_bytes = layout
            .segments()
            .iter()
            .map(|s| match s {
                Segment::Inline(b) => b.len() as u64,
                _ => 0,
            })
            .sum::<u64>();
        let streams_db_rowid = layout.streams_db_rowid();
        Ok(Arc::new(ResolvedFile {
            layout,
            total_len,
            track_id: track.id,
            content_version: track.content_version,
            // Trust boundary: `backing_path` is taken verbatim from the DB row
            // and later opened with default flags (O_RDONLY, symlinks followed)
            // at the serve sites below. The external-writer store contract
            // treats the DB as semi-trusted, so a buggy/hostile writer could
            // point this at an arbitrary path the mount uid can read. There is
            // deliberately no realpath/containment check here: musefs has no
            // serve-time library-root to contain against, and the audio-bytes
            // invariant (served bytes are byte-identical to the named file)
            // still holds — the served bytes are simply *some* file's, not a
            // guaranteed-intended one. Documented, not enforced (#551).
            backing_path: PathBuf::from(&track.backing_path),
            stamp: BackingStamp::from_track(track),
            mtime_secs: mtime_secs_val,
            last_page: Mutex::new(None),
            cache_bytes,
            streams_db_rowid,
        }))
    }
    /// Current number of cached resolved-file entries.
    pub fn entry_count(&self) -> u64 {
        self.cache.len() as u64
    }
    /// Current resident inline-byte weight.
    pub fn weight_bytes(&self) -> u64 {
        self.cache.weight()
    }
    /// Configured resident-byte budget.
    pub fn budget_bytes(&self) -> u64 {
        self.cache.capacity()
    }
    /// Raw key-hit count (NOT content-version-validated hits — see telemetry docs).
    pub fn raw_hits(&self) -> u64 {
        self.cache.hits()
    }
    /// Raw key-miss count.
    pub fn raw_misses(&self) -> u64 {
        self.cache.misses()
    }
}

pub fn read_at_into<M>(
    resolved: &ResolvedFile,
    db: &Db<M>,
    offset: u64,
    size: u64,
    out: &mut Vec<u8>,
) -> Result<()> {
    if offset >= resolved.total_len || size == 0 {
        return Ok(());
    }
    let needs_file = resolved
        .layout
        .segments()
        .iter()
        .any(|s| matches!(s, Segment::BackingAudio { .. } | Segment::OggAudio { .. }));
    // Open and re-validate the backing fd against the stamp the layout was
    // resolved from (#503): between the resolve-time stat and this open the file
    // can be rename-replaced or rewritten in place, which would otherwise splice
    // bytes from a different/modified file behind the stamped header (or
    // short-read against a stale size). The handle fast path validates per read
    // via `validate_opened_backing`; this stateless fallback must too.
    let file = if needs_file {
        crate::metrics::on_open();
        // Opens the semi-trusted DB path verbatim — see the trust-boundary note
        // on `ResolvedFile::backing_path` in `HeaderCache::build` (#551).
        let f = std::fs::File::open(&resolved.backing_path)
            .map_err(|e| CoreError::backing_io(&resolved.backing_path, e))?;
        let f_meta = f
            .metadata()
            .map_err(|e| CoreError::backing_io(&resolved.backing_path, e))?;
        if BackingStamp::from_metadata(&f_meta) != resolved.stamp {
            return Err(CoreError::BackingChanged(
                resolved.backing_path.to_string_lossy().into_owned(),
            ));
        }
        Some(f)
    } else {
        None
    };

    // DB-rowid segments (binary tags AND art) must be read under one WAL
    // snapshot with a `content_version` recheck so a concurrent rowid-reuse
    // (delete + reinsert reusing a freed rowid) can't splice a wrong blob
    // mid-read (#502). Only the rare rowid-streaming layout pays this cost.
    if resolved.streams_db_rowid {
        db.begin_read()?;
        let res = (|| {
            if db.track_content_version(resolved.track_id)? != resolved.content_version {
                // Stale resolve: the layout no longer matches the live row.
                // Surface a retryable error rather than risk wrong bytes.
                return Err(CoreError::BackingChanged(
                    resolved.backing_path.to_string_lossy().into_owned(),
                ));
            }
            read_with_optional_backing(resolved, db, file.as_ref(), offset, size, out)
        })();
        let _ = db.end_read(); // always release the snapshot
        res
    } else {
        read_with_optional_backing(resolved, db, file.as_ref(), offset, size, out)
    }
}

/// Build the optional `BackingReader` from an already-validated `file` and run
/// the segment-splicing loop. Shared by `read_at_into`'s snapshot and
/// non-snapshot branches so the backing-reader wiring lives in one place.
fn read_with_optional_backing<M>(
    resolved: &ResolvedFile,
    db: &Db<M>,
    file: Option<&std::fs::File>,
    offset: u64,
    size: u64,
    out: &mut Vec<u8>,
) -> Result<()> {
    match file {
        Some(file) => {
            let pool = crate::readahead::ReadAheadPool::new(0);
            let buf =
                std::sync::Arc::new(std::sync::Mutex::new(crate::readahead::ReadAhead::new(0)));
            let backing_len = resolved.stamp.size;
            let epoch = std::sync::atomic::AtomicU64::new(0);
            let br =
                crate::readahead::BackingReader::new(file, &buf, &pool, 0, backing_len, &epoch);
            read_segments_into(resolved, Some(db), Some(&br), offset, size, out)
        }
        None => read_segments_into(resolved, Some(db), None, offset, size, out),
    }
}

/// The distinct user-defined keys in `inputs` that the Vorbis synthesis path
/// drops because they are not valid field names. Pure and unit-tested; the
/// caller logs them so a silently-dropped key is observable. Deduped so a
/// multi-valued bad key warns once, not once per value.
fn invalid_vorbis_keys(inputs: &[musefs_format::TagInput]) -> Vec<&str> {
    let mut seen = HashSet::new();
    inputs
        .iter()
        .map(|t| t.key.as_str())
        .filter(|k| !musefs_format::is_valid_vorbis_key(k))
        .filter(|k| seen.insert(*k))
        .collect()
}

/// Allocating form of `read_at_into` (tests and non-hot-path callers).
pub fn read_at<M>(resolved: &ResolvedFile, db: &Db<M>, offset: u64, size: u64) -> Result<Vec<u8>> {
    let mut out = Vec::new();
    read_at_into(resolved, db, offset, size, &mut out)?;
    Ok(out)
}

/// The single segment-splicing loop. `backing` is `Some` whenever the layout has a
/// `BackingAudio`/`OggAudio` segment (guaranteed by `read_at`/`read_at_with_file`);
/// `db` is `Some` whenever the layout has a DB-backed segment
/// (`ArtImage`/`BinaryTag`/`OggArtSlice`, i.e. `streams_db_rowid`). Both arms treat
/// `None` as a contract violation, so a pure-backing layout can be served without a
/// pooled DB connection at all (#520).
fn read_segments_into<M>(
    resolved: &ResolvedFile,
    db: Option<&Db<M>>,
    backing: Option<&crate::readahead::BackingReader>,
    offset: u64,
    size: u64,
    out: &mut Vec<u8>,
) -> Result<()> {
    if offset >= resolved.total_len || size == 0 {
        return Ok(());
    }
    let end = offset.saturating_add(size).min(resolved.total_len);
    out.reserve(usize_from(end - offset));

    let mut seg_start = 0u64;
    for seg in resolved.layout.segments() {
        let seg_len = seg.len();
        let seg_end = seg_start + seg_len;
        let ov_start = offset.max(seg_start);
        let ov_end = end.min(seg_end);
        if ov_start < ov_end {
            let within = ov_start - seg_start;
            let n = usize_from(ov_end - ov_start);
            match seg {
                Segment::Inline(bytes) => {
                    let w = usize_from(within);
                    out.extend_from_slice(&bytes[w..w + n]);
                }
                Segment::BackingAudio { offset: bo, .. } => {
                    let br = backing.expect("backing segment requires an open backing reader");
                    let start = out.len();
                    out.resize(start + n, 0);
                    br.read_exact_at(&mut out[start..], bo + within)?;
                }
                Segment::ArtImage { art_id, .. } => {
                    let db = db.expect("art segment requires a DB connection");
                    let start = out.len();
                    out.resize(start + n, 0);
                    db.read_art_chunk_into(*art_id, within, &mut out[start..])?;
                    crate::metrics::on_art_chunk();
                }
                Segment::BinaryTag { payload_id, .. } => {
                    let db = db.expect("binary-tag segment requires a DB connection");
                    let start = out.len();
                    out.resize(start + n, 0);
                    db.read_binary_tag_chunk_into(*payload_id, within, &mut out[start..])?;
                    crate::metrics::on_binary_tag_chunk();
                }
                Segment::OggAudio {
                    offset: ao,
                    seq_delta,
                    len,
                } => {
                    let br = backing.expect("ogg-audio segment requires an open backing reader");
                    serve_ogg_window(
                        br,
                        *ao,
                        *len,
                        *seq_delta,
                        within,
                        within + n as u64,
                        &mut *out,
                        Some(&resolved.last_page),
                    )?;
                }
                Segment::OggArtSlice {
                    art_id,
                    offset,
                    base64,
                    art_total,
                    ..
                } => {
                    let db = db.expect("ogg-art segment requires a DB connection");
                    if *base64 {
                        let w =
                            musefs_format::ogg::b64_window(*offset + within, n as u64, *art_total);
                        let raw = db.read_art_chunk(*art_id, w.in_start, usize_from(w.in_len))?;
                        crate::metrics::on_art_chunk();
                        let slice = musefs_format::ogg::encode_b64_slice(&raw, w.skip, n)
                            .ok_or_else(|| {
                                CoreError::BackingChanged(format!(
                                    "art {} shorter than its indexed base64 window",
                                    *art_id
                                ))
                            })?;
                        out.extend_from_slice(&slice);
                    } else {
                        let start = out.len();
                        out.resize(start + n, 0);
                        db.read_art_chunk_into(*art_id, *offset + within, &mut out[start..])?;
                        crate::metrics::on_art_chunk();
                    }
                }
            }
        }
        seg_start = seg_end;
        if seg_start >= end {
            break;
        }
    }
    Ok(())
}

/// Serve into `out` from an already-open backing reader (per-handle path). `db`
/// is `None` for a pure-backing layout (no `streams_db_rowid` segment), letting
/// the caller serve without a pooled DB connection (#520).
pub fn read_at_with_file_into<M>(
    resolved: &ResolvedFile,
    db: Option<&Db<M>>,
    backing: &crate::readahead::BackingReader,
    offset: u64,
    size: u64,
    out: &mut Vec<u8>,
) -> Result<()> {
    read_segments_into(resolved, db, Some(backing), offset, size, out)
}

/// Allocating form of `read_at_with_file_into`.
pub fn read_at_with_file<M>(
    resolved: &ResolvedFile,
    db: &Db<M>,
    backing: &crate::readahead::BackingReader,
    offset: u64,
    size: u64,
) -> Result<Vec<u8>> {
    let mut out = Vec::new();
    read_at_with_file_into(resolved, Some(db), backing, offset, size, &mut out)?;
    Ok(out)
}

#[cfg(test)]
mod ogg_serve_tests {
    use super::*;
    use musefs_format::Segment;
    use musefs_format::ogg::page_test_support::lace_packet_pub;
    use std::io::Write;

    #[test]
    fn read_at_renumbers_audio_and_preserves_payload() {
        // Build a file: 8 header bytes + two audio pages (seq 3,4).
        let (mut audio, _) = lace_packet_pub(0x99, 3, false, 10, &[0xA1u8; 200]);
        let (a2, _) = lace_packet_pub(0x99, 4, false, 20, &vec![0xB2u8; 250]);
        audio.extend_from_slice(&a2);
        let audio_offset = 8u64;
        let mut file_bytes = vec![0xFFu8; usize_from(audio_offset)];
        file_bytes.extend_from_slice(&audio);

        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.opus");
        std::fs::File::create(&path)
            .unwrap()
            .write_all(&file_bytes)
            .unwrap();

        let layout = RegionLayout::validated(vec![
            Segment::Inline(b"HDRBYTES".to_vec()), // 8 inline header bytes
            Segment::OggAudio {
                offset: audio_offset,
                len: audio.len() as u64,
                seq_delta: 1, // 3->4, 4->5
            },
        ])
        .unwrap();
        let total = layout.total_len();
        let resolved = ResolvedFile {
            layout,
            total_len: total,
            track_id: 1,
            content_version: 0,
            backing_path: path.clone(),
            // Stamp the real file so the fallback's backing-fd re-validation
            // (#503) passes; a dummy stamp would now read as a changed backing.
            stamp: BackingStamp::from_metadata(&std::fs::metadata(&path).unwrap()),
            mtime_secs: 0,
            last_page: Mutex::new(None),
            cache_bytes: 8,
            streams_db_rowid: false,
        };

        // Read the whole virtual file; needs a Db only for ArtImage (unused here).
        let db = musefs_db::Db::open_in_memory().unwrap();
        let got = read_at(&resolved, &db, 0, total).unwrap();
        assert_eq!(got.len(), usize_from(total));
        assert_eq!(&got[0..8], b"HDRBYTES");

        // The served audio region must have renumbered seqs (4 and 5) and identical
        // payloads to the source.
        let served_audio = &got[8..];
        let h0 = musefs_format::ogg::parse_page(served_audio, 0).unwrap();
        assert_eq!(h0.seq, 4);
        let p1_off = h0.total_len();
        let h1 = musefs_format::ogg::parse_page(served_audio, p1_off).unwrap();
        assert_eq!(h1.seq, 5);
        // Payload bytes unchanged.
        assert!(
            served_audio[h0.header_len..h0.total_len()]
                .iter()
                .all(|&b| b == 0xA1)
        );
        assert!(
            served_audio[p1_off + h1.header_len..p1_off + h1.total_len()]
                .iter()
                .all(|&b| b == 0xB2)
        );
    }
}

#[cfg(test)]
mod resolve_ogg_tests {
    use super::*;
    use musefs_db::{Db, Format, NewTrack, Tag};
    use musefs_format::ogg::page_test_support::lace_packet_pub;
    use std::io::Write;
    use std::os::unix::fs::MetadataExt;

    fn build_opus_file(path: &std::path::Path) -> (u64, u64) {
        let head = b"OpusHead\x01\x02\x38\x01\x80\xbb\x00\x00\x00\x00\x00".to_vec();
        let mut tags = b"OpusTags".to_vec();
        tags.extend_from_slice(&musefs_format::ogg::page_test_support::vorbis_body_empty());
        let (mut bytes, pages) =
            musefs_format::ogg::page_test_support::build_header_pub(0x1234, &[&head, &tags]);
        let audio_offset = bytes.len() as u64;
        let _ = pages;
        let (audio, _) = lace_packet_pub(0x1234, 2, false, 960, &vec![0x7Eu8; 400]);
        bytes.extend_from_slice(&audio);
        std::fs::File::create(path)
            .unwrap()
            .write_all(&bytes)
            .unwrap();
        (audio_offset, bytes.len() as u64 - audio_offset)
    }

    #[test]
    fn resolves_and_reads_opus_with_identical_audio() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("track.opus");
        let (audio_offset, audio_length) = build_opus_file(&path);
        let original = std::fs::read(&path).unwrap();

        let db = Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        let track_id = db
            .upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Opus,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        db.replace_tags(track_id, &[Tag::new("title", "Telephasic Workshop", 0)])
            .unwrap();

        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, track_id).unwrap();
        let out = read_at(&resolved, &db, 0, resolved.total_len).unwrap();

        // The synthesized audio region (after the regenerated header) must be the
        // original audio pages, byte-identical (seq_delta==0 here since the original
        // OpusTags is also an empty-comment musefs-style header of equal page count).
        let header = musefs_format::ogg::read_header(&out).unwrap();
        let synth_audio = &out[usize_from(header.audio_offset)..];
        assert_eq!(synth_audio, &original[usize_from(audio_offset)..]);

        // Tags were rewritten. `ogg::read_tags` now returns canonical lowercase
        // keys for known Vorbis fields (Tasks 1–6 changed the format layer).
        let tags = musefs_format::ogg::read_tags(&out).unwrap();
        assert!(
            tags.iter()
                .any(|(k, v)| k == "title" && v == "Telephasic Workshop")
        );
    }

    #[test]
    fn invalid_vorbis_keys_reports_distinct_out_of_grammar_keys() {
        use musefs_format::TagInput;
        let inputs = vec![
            TagInput::new("artist", "A"),
            TagInput::new("a=b", "c"),
            TagInput::new("a=b", "d"), // same bad key twice -> reported once
            TagInput::new("title", "S"),
        ];
        // Only the out-of-grammar key, deduped; valid keys are not flagged.
        assert_eq!(invalid_vorbis_keys(&inputs), vec!["a=b"]);
    }

    #[test]
    fn synthesis_drops_invalid_vorbis_key_end_to_end() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("track.opus");
        let (audio_offset, audio_length) = build_opus_file(&path);

        let db = Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        let track_id = db
            .upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Opus,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        // `a=b` passes the DB floor but is not a valid Vorbis field name. Without the
        // fix it would synthesize `A=B=c` and re-parse as key "A", value "B=c".
        db.replace_tags(
            track_id,
            &[
                Tag::new("artist", "Alice", 0),
                Tag::new("a=b", "c", 0),
                Tag::new("title", "Song", 0),
            ],
        )
        .unwrap();

        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, track_id).unwrap();
        let out = read_at(&resolved, &db, 0, resolved.total_len).unwrap();

        let tags = musefs_format::ogg::read_tags(&out).unwrap();
        assert!(tags.iter().any(|(k, v)| k == "artist" && v == "Alice"));
        assert!(tags.iter().any(|(k, v)| k == "title" && v == "Song"));
        assert!(
            !tags.iter().any(|(k, _)| k == "A" || k.contains('=')),
            "the a=b key must be dropped, not synthesized as A=B=c: {tags:?}"
        );
    }

    #[test]
    fn read_at_with_file_matches_read_at() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("track.opus");
        let (audio_offset, audio_length) = build_opus_file(&path);
        let db = Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        let track_id = db
            .upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Opus,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, track_id).unwrap();

        let via_open = read_at(&resolved, &db, 0, resolved.total_len).unwrap();
        let file = std::fs::File::open(&resolved.backing_path).unwrap();
        let pool = crate::readahead::ReadAheadPool::new(0);
        let buf = Arc::new(Mutex::new(crate::readahead::ReadAhead::new(0)));
        let epoch = std::sync::atomic::AtomicU64::new(0);
        let br = crate::readahead::BackingReader::new(&file, &buf, &pool, 0, meta.len(), &epoch);
        let via_file = read_at_with_file(&resolved, &db, &br, 0, resolved.total_len).unwrap();
        assert_eq!(via_open, via_file);
    }

    fn build_wav_file(path: &std::path::Path) -> (u64, u64, Vec<u8>) {
        use std::io::Write;
        let mut fmt = Vec::new();
        fmt.extend_from_slice(&1u16.to_le_bytes());
        fmt.extend_from_slice(&1u16.to_le_bytes());
        fmt.extend_from_slice(&44_100u32.to_le_bytes());
        fmt.extend_from_slice(&88_200u32.to_le_bytes());
        fmt.extend_from_slice(&2u16.to_le_bytes());
        fmt.extend_from_slice(&16u16.to_le_bytes());

        let data: Vec<u8> = (0..32u8).collect();
        let mut body = Vec::new();
        for (id, payload) in [(&b"fmt "[..], &fmt[..]), (&b"data"[..], &data[..])] {
            body.extend_from_slice(id);
            body.extend_from_slice(&u32::try_from(payload.len()).unwrap().to_le_bytes());
            body.extend_from_slice(payload);
        }
        let mut bytes = b"RIFF".to_vec();
        bytes.extend_from_slice(&u32::try_from(body.len() + 4).unwrap().to_le_bytes());
        bytes.extend_from_slice(b"WAVE");
        bytes.extend_from_slice(&body);

        let audio_offset = (bytes.len() - data.len()) as u64;
        std::fs::File::create(path)
            .unwrap()
            .write_all(&bytes)
            .unwrap();
        (audio_offset, data.len() as u64, data)
    }

    #[test]
    fn resolves_and_reads_wav_with_identical_audio() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("track.wav");
        let (audio_offset, audio_length, original_data) = build_wav_file(&path);

        let db = Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        let track_id = db
            .upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Wav,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        db.replace_tags(track_id, &[Tag::new("title", "Wave One", 0)])
            .unwrap();

        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, track_id).unwrap();
        let out = read_at(&resolved, &db, 0, resolved.total_len).unwrap();

        // The synthesized output is a valid WAV; its data payload is byte-identical
        // to the original audio.
        let bounds = musefs_format::wav::locate_audio(&out).unwrap();
        assert_eq!(
            &out[usize_from(bounds.audio_offset)
                ..usize_from(bounds.audio_offset + bounds.audio_length)],
            original_data.as_slice()
        );

        // The title was synthesized into the embedded id3 chunk.
        let tags = musefs_format::wav::read_tags(&out);
        assert!(tags.contains(&("title".to_string(), "Wave One".to_string())));
    }

    #[test]
    fn build_cache_bytes_counts_inline_segments_for_ogg() {
        use musefs_db::{Format, NewTrack};
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.opus");
        let (audio_offset, audio_length) = build_opus_file(&path);
        let db = musefs_db::Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        let id = db
            .upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Opus,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, id).unwrap();
        let inline_sum: u64 = resolved
            .layout
            .segments()
            .iter()
            .map(|s| match s {
                Segment::Inline(b) => b.len() as u64,
                _ => 0,
            })
            .sum();
        // SP4: no per-file index estimate; cache_bytes == inline segment bytes only.
        assert_eq!(resolved.cache_bytes, inline_sum);
        assert!(
            inline_sum > 0,
            "Opus header should have non-empty inline segments"
        );
    }
}

#[cfg(test)]
mod ogg_art_serve_tests {
    use super::*;

    #[test]
    fn read_at_serves_base64_art_slice_matching_full_encode() {
        let image: Vec<u8> = (0..1000u32).map(|i| (i % 251) as u8).collect();
        // Compute full base64 via the format crate (base64 is not a direct dep of musefs-core).
        let full_b64 = musefs_format::ogg::encode_b64_slice(
            &image,
            0,
            usize_from(musefs_format::ogg::b64_len(image.len() as u64)),
        )
        .expect("full-length window lies within the encoded output");

        let db = musefs_db::Db::open_in_memory().unwrap();
        let art_id = db
            .upsert_art(&musefs_db::NewArt {
                mime: "image/png".to_string(),
                width: Some(1),
                height: Some(1),
                data: image.clone(),
            })
            .unwrap();

        let layout = RegionLayout::validated(vec![
            Segment::Inline(b"HEAD".to_vec()),
            Segment::OggArtSlice {
                art_id,
                offset: 0,
                len: musefs_format::BlobLen::new(full_b64.len() as u64).unwrap(),
                base64: true,
                art_total: image.len() as u64,
            },
            Segment::Inline(b"XY".to_vec()),
        ])
        .unwrap();
        let total = layout.total_len();
        let resolved = ResolvedFile {
            layout,
            total_len: total,
            track_id: 1,
            content_version: 0,
            backing_path: std::path::PathBuf::from("/dev/null"),
            stamp: BackingStamp {
                size: 0,
                mtime_ns: 0,
                ctime_ns: 0,
            },
            mtime_secs: 0,
            last_page: Mutex::new(None),
            cache_bytes: 0,
            streams_db_rowid: false,
        };

        // Full read.
        let got = read_at(&resolved, &db, 0, total).unwrap();
        let mut want = b"HEAD".to_vec();
        want.extend_from_slice(&full_b64);
        want.extend_from_slice(b"XY");
        assert_eq!(got, want);

        // Partial read straddling into the middle of the art slice (non-4-aligned).
        let part = read_at(&resolved, &db, 7, 23).unwrap();
        assert_eq!(part, want[7..30]);
    }

    #[test]
    fn read_at_serves_raw_art_slice() {
        let image: Vec<u8> = (0..300u32)
            .map(|i| u8::try_from(i % 256).unwrap())
            .collect();
        let db = musefs_db::Db::open_in_memory().unwrap();
        let art_id = db
            .upsert_art(&musefs_db::NewArt {
                mime: "image/png".to_string(),
                width: None,
                height: None,
                data: image.clone(),
            })
            .unwrap();
        let layout = RegionLayout::validated(vec![Segment::OggArtSlice {
            art_id,
            offset: 0,
            len: musefs_format::BlobLen::new(image.len() as u64).unwrap(),
            base64: false,
            art_total: image.len() as u64,
        }])
        .unwrap();
        let total = layout.total_len();
        let resolved = ResolvedFile {
            layout,
            total_len: total,
            track_id: 1,
            content_version: 0,
            backing_path: std::path::PathBuf::from("/dev/null"),
            stamp: BackingStamp {
                size: 0,
                mtime_ns: 0,
                ctime_ns: 0,
            },
            mtime_secs: 0,
            last_page: Mutex::new(None),
            cache_bytes: 0,
            streams_db_rowid: false,
        };
        let got = read_at(&resolved, &db, 10, 50).unwrap();
        assert_eq!(got, image[10..60]);
    }
}

#[cfg(test)]
mod cache_bound_tests {
    use super::*;
    use musefs_db::{Db, Format, NewTrack};
    use std::os::unix::fs::MetadataExt;

    #[test]
    fn header_cache_exposes_budget_and_starts_empty() {
        let c = HeaderCache::with_budget(Mode::Synthesis, 1234);
        assert_eq!(c.entry_count(), 0);
        assert_eq!(c.weight_bytes(), 0);
        assert!(
            c.budget_bytes() >= 1234,
            "budget must be at least the requested amount"
        );
    }

    #[test]
    fn header_cache_counts_entries_weight_hits_and_misses() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_in_memory().unwrap();
        let mut ids = Vec::new();
        for name in ["a.flac", "b.flac"] {
            let path = dir.path().join(name);
            let (audio_offset, audio_length) = write_flac_local(&path);
            let meta = std::fs::metadata(&path).unwrap();
            ids.push(
                db.upsert_track(&NewTrack {
                    backing_path: path.to_string_lossy().into_owned(),
                    format: Format::Flac,
                    audio_offset,
                    audio_length,
                    backing_size: meta.len(),
                    backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                    backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
                })
                .unwrap(),
            );
        }
        let cache = HeaderCache::new(Mode::Synthesis);
        for id in &ids {
            cache.resolve(&db, *id).unwrap(); // miss → build + insert
            cache.resolve(&db, *id).unwrap(); // hit (content_version unchanged)
        }
        assert_eq!(cache.entry_count(), 2);
        assert_eq!(cache.raw_hits(), 2);
        assert_eq!(cache.raw_misses(), 2);
        assert!(
            cache.weight_bytes() > 0,
            "synthesis entries carry inline header bytes"
        );
    }

    fn entry(content_version: i64, inline_len: usize) -> Arc<ResolvedFile> {
        Arc::new(ResolvedFile {
            layout: RegionLayout::new_unchecked(vec![Segment::Inline(vec![0u8; inline_len])]),
            total_len: inline_len as u64,
            track_id: 1,
            content_version,
            backing_path: std::path::PathBuf::from("/nonexistent"),
            stamp: BackingStamp {
                size: 0,
                mtime_ns: 0,
                ctime_ns: 0,
            },
            mtime_secs: 0,
            last_page: Mutex::new(None),
            cache_bytes: inline_len as u64,
            streams_db_rowid: false,
        })
    }

    #[test]
    fn header_cache_resolve_caches_by_content_version() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.flac");
        let (audio_offset, audio_length) = write_flac_local(&path);
        let db = Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        let id = db
            .upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Flac,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        let cache = HeaderCache::new(Mode::Synthesis); // NOTE: not `mut` — resolve is &self now
        let a = cache.resolve(&db, id).unwrap();
        let b = cache.resolve(&db, id).unwrap();
        assert!(Arc::ptr_eq(&a, &b));
    }

    #[test]
    fn resolve_is_safe_under_concurrent_access() {
        // Many threads resolving the same track exercise the off-lock build race
        // (concurrent miss → build → insert on one shard) and concurrent gets.
        // Each thread needs its own connection (Db is !Sync), so use a file-backed
        // DB and open_readonly per thread.
        let dir = tempfile::tempdir().unwrap();
        let flac_path = dir.path().join("a.flac");
        let (audio_offset, audio_length) = write_flac_local(&flac_path);
        let db_path = dir.path().join("m.db");
        let track_id = {
            let db = Db::open(&db_path).unwrap();
            let meta = std::fs::metadata(&flac_path).unwrap();
            db.upsert_track(&NewTrack {
                backing_path: flac_path.to_string_lossy().into_owned(),
                format: Format::Flac,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap()
        };

        let cache = std::sync::Arc::new(HeaderCache::new(Mode::Synthesis));
        std::thread::scope(|s| {
            for _ in 0..8 {
                let cache = std::sync::Arc::clone(&cache);
                let db_path = db_path.clone();
                s.spawn(move || {
                    let db = Db::open_readonly(&db_path).unwrap();
                    for _ in 0..50 {
                        let r = cache.resolve(&db, track_id).unwrap();
                        assert!(r.total_len > 0);
                        assert_eq!(r.content_version, 0);
                    }
                });
            }
        });
    }

    #[test]
    fn header_cache_retain_drops_absent_tracks() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_in_memory().unwrap();
        let mk = |name: &str| {
            let path = dir.path().join(name);
            let (audio_offset, audio_length) = write_flac_local(&path);
            let meta = std::fs::metadata(&path).unwrap();
            db.upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Flac,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap()
        };
        let keep = mk("keep.flac");
        let gone = mk("gone.flac");
        let cache = HeaderCache::new(Mode::Synthesis);
        let keep_a = cache.resolve(&db, keep).unwrap();
        let gone_a = cache.resolve(&db, gone).unwrap();

        let live: HashSet<i64> = [keep].into_iter().collect();
        cache.retain(&live);

        // The kept track stays the same cached Arc; the dropped one re-resolves fresh.
        assert!(Arc::ptr_eq(&keep_a, &cache.resolve(&db, keep).unwrap()));
        assert!(!Arc::ptr_eq(&gone_a, &cache.resolve(&db, gone).unwrap()));
    }

    #[test]
    fn header_cache_remove_drops_one_track_only() {
        let dir = tempfile::tempdir().unwrap();
        let db = Db::open_in_memory().unwrap();
        let mk = |name: &str| {
            let path = dir.path().join(name);
            let (audio_offset, audio_length) = write_flac_local(&path);
            let meta = std::fs::metadata(&path).unwrap();
            db.upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Flac,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap()
        };
        let keep = mk("keep.flac");
        let gone = mk("gone.flac");
        let cache = HeaderCache::new(Mode::Synthesis);
        let keep_a = cache.resolve(&db, keep).unwrap();
        let gone_a = cache.resolve(&db, gone).unwrap();

        cache.remove(gone);

        // The kept track stays the same cached Arc; the removed one re-resolves fresh.
        assert!(Arc::ptr_eq(&keep_a, &cache.resolve(&db, keep).unwrap()));
        assert!(!Arc::ptr_eq(&gone_a, &cache.resolve(&db, gone).unwrap()));
    }

    #[test]
    fn default_cache_budget_is_64_mib() {
        assert_eq!(DEFAULT_CACHE_BUDGET, 67_108_864);
    }

    #[test]
    fn read_segments_returns_empty_past_end_of_range() {
        let db = musefs_db::Db::open_in_memory().unwrap();
        let resolved = entry(0, 10);
        let out = read_at(&resolved, &db, 11, 1).unwrap();
        assert!(out.is_empty());
        let out0 = read_at(&resolved, &db, 0, 0).unwrap();
        assert!(out0.is_empty());
    }

    fn track_with_bounds(
        path: &std::path::Path,
        audio_offset: u64,
        audio_length: u64,
    ) -> (musefs_db::Db, i64) {
        use musefs_db::{Format, NewTrack};
        let db = musefs_db::Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(path).unwrap();
        let id = db
            .upsert_track(&NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: Format::Flac,
                audio_offset,
                audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        (db, id)
    }

    #[test]
    fn build_rejects_audio_region_past_end_of_file() {
        // An audio region past the end of the backing file (offset + length >
        // backing_size) is rejected at write time by the V4 bounds CHECK — it can
        // no longer be committed and reach synthesis.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.flac");
        let _ = write_flac_local(&path);
        let meta = std::fs::metadata(&path).unwrap();
        let db = musefs_db::Db::open_in_memory().unwrap();
        let rejected = db.upsert_track(&musefs_db::NewTrack {
            backing_path: path.to_string_lossy().into_owned(),
            format: musefs_db::Format::Flac,
            audio_offset: meta.len(),
            audio_length: 5,
            backing_size: meta.len(),
            backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
            backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
        });
        assert!(
            rejected.is_err(),
            "bounds CHECK must reject an over-EOF audio region"
        );
    }

    #[test]
    fn build_accepts_audio_region_ending_exactly_at_eof() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.flac");
        let (audio_offset, audio_length) = write_flac_local(&path);
        let (db, id) = track_with_bounds(&path, audio_offset, audio_length);
        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache
            .resolve(&db, id)
            .expect("exact-fit bounds must resolve");
        assert!(resolved.total_len > 0);
    }

    #[test]
    fn build_accepts_audio_region_ending_before_eof() {
        // A valid track whose audio region ends strictly before EOF
        // (audio_offset + audio_length < backing_size, allowed by TrackBounds)
        // must still resolve: the bounds guard rejects only an over-EOF region.
        // Pins the guard's `>` against `<`, which would spuriously reject every
        // sub-EOF track.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.flac");
        let (audio_offset, audio_length) = write_flac_local(&path);
        // Append trailing bytes so the audio region no longer reaches EOF; the
        // padded length becomes backing_size, leaving offset + length < it.
        use std::io::Write;
        std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap()
            .write_all(&[0u8; 64])
            .unwrap();
        let (db, id) = track_with_bounds(&path, audio_offset, audio_length);
        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, id).expect("sub-EOF bounds must resolve");
        assert!(resolved.total_len > 0);
    }

    #[test]
    fn build_cache_bytes_counts_inline_segments() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.flac");
        let (audio_offset, audio_length) = write_flac_local(&path);
        let (db, id) = track_with_bounds(&path, audio_offset, audio_length);
        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, id).unwrap();
        let inline_sum: u64 = resolved
            .layout
            .segments()
            .iter()
            .map(|s| match s {
                Segment::Inline(b) => b.len() as u64,
                _ => 0,
            })
            .sum();
        assert!(inline_sum > 0);
        assert_eq!(resolved.cache_bytes, inline_sum);
    }

    #[test]
    fn build_rejects_layout_failing_validation() {
        // A layout with an empty Inline segment fails validate(); the defensive
        // check at the cache boundary must surface it rather than cache it.
        let bad = RegionLayout::new_unchecked(vec![Segment::Inline(vec![])]);
        let err = bad.validate();
        assert!(err.is_err());
    }

    fn write_flac_local(path: &std::path::Path) -> (u64, u64) {
        fn block(bt: u8, body: &[u8], last: bool) -> Vec<u8> {
            let mut v = vec![(if last { 0x80 } else { 0 }) | (bt & 0x7F)];
            let n: u32 = u32::try_from(body.len()).unwrap();
            v.extend_from_slice(&[
                u8::try_from(n >> 16).unwrap(),
                u8::try_from(n >> 8).unwrap(),
                u8::try_from(n).unwrap(),
            ]);
            v.extend_from_slice(body);
            v
        }
        let mut si = vec![
            0x10, 0x00, 0x10, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x0A, 0xC4, 0x42, 0xF0,
            0x00, 0x00, 0x00, 0x00,
        ];
        si.extend_from_slice(&[0u8; 16]);
        let mut vc = Vec::new();
        let vendor = b"x";
        vc.extend_from_slice(&u32::try_from(vendor.len()).unwrap().to_le_bytes());
        vc.extend_from_slice(vendor);
        vc.extend_from_slice(&0u32.to_le_bytes());
        let mut out = b"fLaC".to_vec();
        out.extend(block(0, &si, false));
        out.extend(block(4, &vc, true));
        let audio = [0xABu8; 256];
        let audio_offset = out.len() as u64;
        out.extend_from_slice(&audio);
        std::fs::write(path, &out).unwrap();
        (audio_offset, audio.len() as u64)
    }

    #[test]
    fn cache_weight_stays_within_budget_after_flood() {
        let cache = HeaderCache::with_budget(Mode::Synthesis, 4096);
        for id in 0..64i64 {
            cache.cache.insert(id, entry(0, 256)); // 64 × 256 B = 16 KiB ≫ 4 KiB
        }
        // End-state assertion only: quick_cache does not document per-insert
        // synchronous eviction, so the per-insert bound is not guaranteed.
        assert!(
            cache.cache.weight() <= 4096,
            "total weight {} exceeds the 4096-byte budget",
            cache.cache.weight()
        );
        // len() is assumed to count resident entries. If this assertion ever
        // trips, the diagnosis is the same as the weight() note above: re-read
        // the spec's eviction-timing section and escalate — don't loosen.
        assert!(
            cache.cache.len() < 64,
            "no eviction happened: all 64 over-budget entries are resident"
        );
    }

    #[test]
    fn zero_cache_bytes_entry_still_weighs_one() {
        // StructureOnly layouts have cache_bytes == 0; the weigher's .max(1) keeps
        // them inside the weighted bound instead of escaping it (quick_cache
        // ignores zero-weight entries when evicting).
        let cache = HeaderCache::with_budget(Mode::StructureOnly, 1024);
        cache.cache.insert(1, entry(0, 0));
        assert_eq!(cache.cache.weight(), 1);
        assert!(cache.cache.get(&1).is_some());
    }
}

#[cfg(test)]
mod binary_tag_serve_tests {
    use super::*;
    use musefs_db::{BinaryTag, NewTrack};
    use std::os::unix::fs::MetadataExt;

    #[test]
    fn resolve_mp3_emits_binary_tag_in_synthesized_region() {
        use id3::frame::{Content, Unknown};
        use id3::{Encoder, Frame, Tag, TagLike, Version};
        let dir = tempfile::tempdir().unwrap();
        let mut tag = Tag::new();
        let needle = [0xDE, 0xAD, 0xBE, 0xEF, 0x01, 0x02, 0x77, 0x88];
        tag.add_frame(Frame::with_content(
            "PRIV",
            Content::Unknown(Unknown {
                data: needle.to_vec(),
                version: Version::Id3v24,
            }),
        ));
        let mut bytes = Vec::new();
        Encoder::new()
            .version(Version::Id3v24)
            .encode(&tag, &mut bytes)
            .unwrap();
        bytes.extend_from_slice(&[0xFF, 0xFB, 0x90, 0x00]);
        let path = dir.path().join("a.mp3");
        std::fs::write(&path, &bytes).unwrap();

        let db = musefs_db::Db::open_in_memory().unwrap();
        let bounds = musefs_format::mp3::locate_audio(&bytes).unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        let tid = db
            .upsert_track(&musefs_db::NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: musefs_db::Format::Mp3,
                audio_offset: bounds.audio_offset,
                audio_length: bounds.audio_length,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        db.set_binary_tags(
            tid,
            &[musefs_db::BinaryTag {
                key: "PRIV".into(),
                payload: needle.to_vec(),
                ordinal: 0,
            }],
        )
        .unwrap();

        let cache = crate::reader::HeaderCache::new(crate::Mode::Synthesis);
        let resolved = cache.resolve(&db, tid).unwrap();
        let whole = crate::reader::read_at(&resolved, &db, 0, resolved.total_len).unwrap();
        assert!(
            whole.windows(needle.len()).any(|w| w == needle),
            "PRIV body not in synthesized file"
        );
    }

    #[test]
    fn read_at_serves_binary_tag_segment() {
        let db = Db::open_in_memory().unwrap();
        let id = db
            .upsert_track(&NewTrack {
                backing_path: "/x.mp3".into(),
                format: Format::Mp3,
                audio_offset: 0,
                audio_length: 0,
                backing_size: 0,
                backing_mtime_ns: 0,
                backing_ctime_ns: 0,
            })
            .unwrap();
        db.set_binary_tags(
            id,
            &[BinaryTag {
                key: "PRIV".into(),
                payload: vec![10, 20, 30, 40],
                ordinal: 0,
            }],
        )
        .unwrap();
        let rowid = db.get_binary_tags(id).unwrap()[0].rowid;

        let resolved = ResolvedFile {
            layout: RegionLayout::validated(vec![Segment::BinaryTag {
                payload_id: rowid,
                len: musefs_format::BlobLen::new(4).unwrap(),
            }])
            .unwrap(),
            total_len: 4,
            track_id: id,
            // Match the live row: set_binary_tags bumps content_version via
            // trigger, and the rowid-streaming read path rechecks it (#502).
            content_version: db.track_content_version(id).unwrap(),
            backing_path: PathBuf::from("/x.mp3"),
            stamp: BackingStamp {
                size: 0,
                mtime_ns: 0,
                ctime_ns: 0,
            },
            mtime_secs: 0,
            last_page: Mutex::new(None),
            cache_bytes: 0,
            streams_db_rowid: true,
        };
        // No BackingAudio segment, so read_at opens no file.
        let got = read_at(&resolved, &db, 1, 2).unwrap();
        assert_eq!(got, vec![20, 30]);
    }

    #[test]
    fn fallback_read_rejects_changed_backing() {
        // #503: the stateless read path must re-validate the freshly opened
        // backing fd against the resolved stamp, like the handle fast path does.
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("a.mp3");
        std::fs::write(&path, vec![0u8; 100]).unwrap();
        let db = Db::open_in_memory().unwrap();
        let layout = RegionLayout::validated(vec![
            Segment::Inline(vec![1, 2, 3]),
            Segment::BackingAudio {
                offset: 0,
                len: 100,
            },
        ])
        .unwrap();
        let total = layout.total_len();
        let resolved = ResolvedFile {
            layout,
            total_len: total,
            track_id: 1,
            content_version: 0,
            backing_path: path.clone(),
            stamp: BackingStamp::from_metadata(&std::fs::metadata(&path).unwrap()),
            mtime_secs: 0,
            last_page: Mutex::new(None),
            cache_bytes: 3,
            streams_db_rowid: false,
        };
        // Matching stamp: the read succeeds.
        assert!(read_at(&resolved, &db, 0, total).is_ok());
        // Replace the backing file with a different size -> stamp mismatch.
        std::fs::write(&path, vec![0u8; 200]).unwrap();
        let err = read_at(&resolved, &db, 0, total).unwrap_err();
        assert!(matches!(err, CoreError::BackingChanged(_)), "{err:?}");
    }

    #[test]
    fn fallback_read_of_art_rechecks_content_version() {
        // #502: an art-only layout (no BinaryTag) must take the snapshot +
        // content_version recheck path on the stateless fallback, so a stale
        // resolve cannot stream a reused art rowid's bytes.
        let db = Db::open_in_memory().unwrap();
        let id = db
            .upsert_track(&NewTrack {
                backing_path: "/y.mp3".into(),
                format: Format::Mp3,
                audio_offset: 0,
                audio_length: 0,
                backing_size: 0,
                backing_mtime_ns: 0,
                backing_ctime_ns: 0,
            })
            .unwrap();
        let art_id = db
            .upsert_art(&musefs_db::NewArt {
                mime: "image/png".into(),
                width: None,
                height: None,
                data: vec![1, 2, 3, 4],
            })
            .unwrap();
        let layout = RegionLayout::validated(vec![Segment::ArtImage {
            art_id,
            len: musefs_format::BlobLen::new(4).unwrap(),
        }])
        .unwrap();
        let live_cv = db.track_content_version(id).unwrap();
        let mk = |content_version| ResolvedFile {
            layout: layout.clone(),
            total_len: 4,
            track_id: id,
            content_version,
            backing_path: PathBuf::from("/y.mp3"),
            stamp: BackingStamp {
                size: 0,
                mtime_ns: 0,
                ctime_ns: 0,
            },
            mtime_secs: 0,
            last_page: Mutex::new(None),
            cache_bytes: 0,
            streams_db_rowid: true,
        };
        // Live content_version: art bytes are served.
        assert_eq!(read_at(&mk(live_cv), &db, 0, 4).unwrap(), vec![1, 2, 3, 4]);
        // Stale content_version: the recheck (now covering art) rejects.
        let err = read_at(&mk(live_cv + 1), &db, 0, 4).unwrap_err();
        assert!(matches!(err, CoreError::BackingChanged(_)), "{err:?}");
    }
}

#[cfg(test)]
mod serve_cap_tests {
    use super::*;
    use musefs_db::{Db, Format, NewTrack};

    const CAP: u64 = crate::scan::MAX_PROBE_BYTES;

    /// A sparse backing file of `len` bytes (no real bytes written — `set_len`
    /// only extends the file's logical size, which tmpfs keeps sparse).
    fn sparse_file(dir: &std::path::Path, name: &str, len: u64) -> std::path::PathBuf {
        let path = dir.join(name);
        let f = std::fs::File::create(&path).unwrap();
        f.set_len(len).unwrap();
        path
    }

    /// Insert a `tracks` row whose `audio_offset` exceeds the cap while still
    /// satisfying both serve guards (`backing_size == meta.len()` and
    /// `audio_offset + audio_length <= meta.len()`). Returns the track id.
    /// Takes `&Db` (= `Db<ReadWrite>`) because `upsert_track` is defined on
    /// `impl Db<ReadWrite>`, not the generic `impl<M> Db<M>`.
    fn hostile_track(db: &Db, path: &std::path::Path, format: Format) -> i64 {
        let meta = std::fs::metadata(path).unwrap();
        let stamp = BackingStamp::from_metadata(&meta);
        db.upsert_track(&NewTrack {
            backing_path: path.to_string_lossy().into_owned(),
            format,
            audio_offset: CAP + 1,
            audio_length: 1,
            backing_size: meta.len(),
            backing_mtime_ns: stamp.mtime_ns,
            backing_ctime_ns: stamp.ctime_ns,
        })
        .unwrap()
    }

    /// Assert a resolve attempt fails closed with the cap error for `audio_offset`.
    fn assert_capped(result: crate::Result<std::sync::Arc<ResolvedFile>>) {
        match result {
            Err(CoreError::HeaderTooLarge { requested, cap }) => {
                assert_eq!(requested, CAP + 1);
                assert_eq!(cap, CAP);
            }
            Err(other) => panic!("expected HeaderTooLarge, got {other:?}"),
            Ok(_) => panic!("expected HeaderTooLarge, resolve unexpectedly succeeded"),
        }
    }

    #[test]
    fn wav_serve_caps_hostile_offset() {
        let dir = tempfile::tempdir().unwrap();
        let path = sparse_file(dir.path(), "hostile.wav", CAP + 2);
        let db = Db::open_in_memory().unwrap();
        let track_id = hostile_track(&db, &path, Format::Wav);

        let cache = HeaderCache::new(Mode::Synthesis);
        assert_capped(cache.resolve(&db, track_id));
    }

    #[test]
    fn ogg_serve_caps_hostile_offset() {
        let dir = tempfile::tempdir().unwrap();
        let path = sparse_file(dir.path(), "hostile.opus", CAP + 2);
        let db = Db::open_in_memory().unwrap();
        let track_id = hostile_track(&db, &path, Format::Opus);

        let cache = HeaderCache::new(Mode::Synthesis);
        assert_capped(cache.resolve(&db, track_id));
    }

    #[test]
    fn flac_legacy_serve_caps_hostile_offset() {
        let dir = tempfile::tempdir().unwrap();
        let path = sparse_file(dir.path(), "hostile.flac", CAP + 2);
        let db = Db::open_in_memory().unwrap();
        // No structural-block rows inserted -> build() takes the legacy fallback
        // branch (rows.is_empty()) that calls read_front.
        let track_id = hostile_track(&db, &path, Format::Flac);
        assert!(db.get_structural_blocks(track_id).unwrap().is_empty());

        let cache = HeaderCache::new(Mode::Synthesis);
        assert_capped(cache.resolve(&db, track_id));
    }

    #[test]
    fn read_front_rejects_oversize_before_open() {
        // Nonexistent path: if the cap check did NOT fire first, File::open would
        // error and we'd get an Io error instead of HeaderTooLarge. So this also
        // pins the fail-closed ordering (check precedes any open/allocation).
        let err =
            read_front(std::path::Path::new("/nonexistent/musefs/front"), CAP + 1).unwrap_err();
        match err {
            CoreError::HeaderTooLarge { requested, cap } => {
                assert_eq!(requested, CAP + 1);
                assert_eq!(cap, CAP);
            }
            other => panic!("expected HeaderTooLarge, got {other:?}"),
        }
    }

    #[test]
    fn read_front_allows_exactly_cap() {
        // Boundary: `n == CAP` must NOT be rejected — the check is `>`, not `>=`.
        // With a nonexistent path the call still fails, but with an Io error from
        // File::open, never HeaderTooLarge. This pins the boundary and kills the
        // `> -> >=` mutant.
        let err = read_front(std::path::Path::new("/nonexistent/musefs/front"), CAP).unwrap_err();
        assert!(
            matches!(err, CoreError::BackingIo { .. }),
            "expected a backing-file Io error at the cap boundary, got {err:?}"
        );
    }

    #[test]
    fn read_front_io_error_carries_the_backing_path() {
        // The most common passthrough failure (a moved/inaccessible backing file)
        // must name the path rather than collapse to a pathless `Io`/EIO (#521).
        let p = std::path::Path::new("/nonexistent/musefs/backing.flac");
        match read_front(p, 16).unwrap_err() {
            CoreError::BackingIo { path, .. } => assert_eq!(path, p),
            other => panic!("expected BackingIo carrying the path, got {other:?}"),
        }
    }
}

#[cfg(test)]
mod readahead_differential_tests {
    use super::*;
    use crate::readahead::{BackingReader, ReadAhead, ReadAheadPool};
    use std::sync::{Arc, Mutex};

    fn pcm_fixture() -> (musefs_db::Db, Arc<ResolvedFile>, std::fs::File) {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("test.wav");
        let mut body = Vec::new();
        body.extend_from_slice(b"fmt ");
        body.extend_from_slice(&16u32.to_le_bytes());
        body.extend_from_slice(&1u16.to_le_bytes());
        body.extend_from_slice(&1u16.to_le_bytes());
        body.extend_from_slice(&44100u32.to_le_bytes());
        body.extend_from_slice(&88200u32.to_le_bytes());
        body.extend_from_slice(&2u16.to_le_bytes());
        body.extend_from_slice(&16u16.to_le_bytes());
        let audio_data: Vec<u8> = (0..1024u32).map(|i| (i % 251) as u8).collect();
        body.extend_from_slice(b"data");
        body.extend_from_slice(&u32::try_from(audio_data.len()).unwrap().to_le_bytes());
        body.extend_from_slice(&audio_data);
        let mut riff = b"RIFF".to_vec();
        riff.extend_from_slice(&u32::try_from(body.len()).unwrap().to_le_bytes());
        riff.extend_from_slice(b"WAVE");
        riff.extend_from_slice(&body);
        let audio_offset = (riff.len() - audio_data.len()) as u64;
        std::fs::write(&path, &riff).unwrap();

        let db = musefs_db::Db::open_in_memory().unwrap();
        let meta = std::fs::metadata(&path).unwrap();
        use std::os::unix::fs::MetadataExt;
        let track_id = db
            .upsert_track(&musefs_db::NewTrack {
                backing_path: path.to_string_lossy().into_owned(),
                format: musefs_db::Format::Wav,
                audio_offset,
                audio_length: audio_data.len() as u64,
                backing_size: meta.len(),
                backing_mtime_ns: meta.mtime() * 1_000_000_000 + meta.mtime_nsec(),
                backing_ctime_ns: meta.ctime() * 1_000_000_000 + meta.ctime_nsec(),
            })
            .unwrap();
        let cache = HeaderCache::new(Mode::Synthesis);
        let resolved = cache.resolve(&db, track_id).unwrap();
        let file = std::fs::File::open(&resolved.backing_path).unwrap();
        (db, resolved, file)
    }

    fn oracle_read(
        resolved: &ResolvedFile,
        file: &std::fs::File,
        offset: u64,
        size: u64,
        out: &mut Vec<u8>,
    ) -> Result<()> {
        if offset >= resolved.total_len || size == 0 {
            return Ok(());
        }
        let end = offset.saturating_add(size).min(resolved.total_len);
        out.reserve(usize_from(end - offset));
        let mut seg_start = 0u64;
        for seg in resolved.layout.segments() {
            let seg_len = seg.len();
            let seg_end = seg_start + seg_len;
            let ov_start = offset.max(seg_start);
            let ov_end = end.min(seg_end);
            if ov_start < ov_end {
                let within = ov_start - seg_start;
                let n = usize_from(ov_end - ov_start);
                match seg {
                    Segment::Inline(bytes) => {
                        let w = usize_from(within);
                        out.extend_from_slice(&bytes[w..w + n]);
                    }
                    Segment::BackingAudio { offset: bo, .. } => {
                        let start = out.len();
                        out.resize(start + n, 0);
                        use std::os::unix::fs::FileExt;
                        file.read_exact_at(&mut out[start..], bo + within)?;
                    }
                    _ => panic!("unexpected segment in PCM fixture"),
                }
            }
            seg_start = seg_end;
            if seg_start >= end {
                break;
            }
        }
        Ok(())
    }

    #[test]
    fn pcm_bytes_identical_through_backing_reader() {
        let (db, resolved, file) = pcm_fixture();
        let pool = ReadAheadPool::new(0);
        let buf = Arc::new(Mutex::new(ReadAhead::new(0)));
        let epoch = std::sync::atomic::AtomicU64::new(0);
        let br = BackingReader::new(&file, &buf, &pool, 0, resolved.stamp.size, &epoch);
        let total = resolved.total_len;
        for &size in &[1u64, 7, 4096, 65536, 262_144] {
            let mut off = 0;
            while off < total {
                let n = size.min(total - off);
                let mut via = Vec::new();
                read_segments_into(&resolved, Some(&db), Some(&br), off, n, &mut via).unwrap();
                let mut direct = Vec::new();
                oracle_read(&resolved, &file, off, n, &mut direct).unwrap();
                assert_eq!(via, direct, "mismatch at off={off} size={size}");
                off += n;
            }
        }
    }

    #[test]
    fn pcm_bytes_identical_under_forced_eviction() {
        let (db, resolved, file) = pcm_fixture();
        let pool = ReadAheadPool::new(1024 * 1024);
        let buf = Arc::new(Mutex::new(ReadAhead::new(pool.per_stream_cap())));
        pool.register(1, Arc::clone(&buf));
        let epoch = std::sync::atomic::AtomicU64::new(0);
        let br = BackingReader::new(&file, &buf, &pool, 1, resolved.stamp.size, &epoch);
        let total = resolved.total_len;
        let mut off = 0;
        while off < total {
            let n = 65536u64.min(total - off);
            let mut via = Vec::new();
            read_segments_into(&resolved, Some(&db), Some(&br), off, n, &mut via).unwrap();
            let mut direct = Vec::new();
            oracle_read(&resolved, &file, off, n, &mut direct).unwrap();
            assert_eq!(via, direct, "eviction mismatch at {off}");
            off += n;
        }
    }

    #[test]
    fn partial_overlap_seek_serves_correct_bytes() {
        let (db, resolved, file) = pcm_fixture();
        let pool = ReadAheadPool::new(64 * 1024 * 1024);
        let buf = Arc::new(Mutex::new(ReadAhead::new(pool.per_stream_cap())));
        pool.register(1, Arc::clone(&buf));
        let epoch = std::sync::atomic::AtomicU64::new(0);
        let br = BackingReader::new(&file, &buf, &pool, 1, resolved.stamp.size, &epoch);
        let seq = [(0u64, 600u64), (590, 50), (10, 4096), (12, 4096)];
        for &(off, n) in &seq {
            let n = n.min(resolved.total_len.saturating_sub(off));
            if n == 0 {
                continue;
            }
            let mut via = Vec::new();
            read_segments_into(&resolved, Some(&db), Some(&br), off, n, &mut via).unwrap();
            let mut direct = Vec::new();
            oracle_read(&resolved, &file, off, n, &mut direct).unwrap();
            assert_eq!(via, direct, "partial-seek mismatch at {off}+{n}");
        }
    }
}