//! The block cache every remote backend is read through.
//!
//! `docs/design/range-reads.md` §2 measured why it is required: the parsers
//! revisit the same structures many times (410 reads to list the 21 objects
//! of a 7.7 MB file), so mapping reads one-to-one onto requests is hopeless,
//! while a cache of 1 MiB blocks lists the same file in 2 requests.
//!
//! [`BlockCache`] wraps any [`Storage`] (the backend that does the fetching)
//! and serves reads from fixed-size, aligned blocks:
//!
//! - **LRU with a byte budget.** Blocks are evicted least-recently-used
//!   first once the cached bytes exceed [`CacheConfig::capacity`]. Reads
//!   hold their blocks by reference count, so eviction never invalidates
//!   data a read is still using.
//! - **Coalescing.** All the blocks one `read_at`/`read_ranges` call misses
//!   are fetched with one `read_ranges` call on the backend, as runs of
//!   consecutive blocks (a gap of up to [`CacheConfig::coalesce_gap`] bytes
//!   of uncached blocks is fetched too, to merge two runs), each run at
//!   most [`CacheConfig::max_request`] bytes. A backend fetches the runs of
//!   one call in parallel.
//! - **Concurrency.** The cache's lock is held only to look blocks up and
//!   to insert them, never across a fetch. A block being fetched is marked
//!   in flight: a second reader that needs it waits for that fetch instead
//!   of issuing its own, so concurrent readers never fetch a block twice.
//!   A failed fetch fails every reader waiting for it, and is not cached.
//! - **Large reads do not flush the cache.** A call whose missing blocks
//!   add up to more than half the budget is served without keeping them
//!   (a big chunked read would otherwise evict all the metadata).
//! - **Local storages pass through.** A backend that holds the whole file
//!   in memory ([`Storage::as_contiguous`]) is read directly.

use std::borrow::Cow;
use std::collections::{BTreeMap, BTreeSet, HashMap};
use std::ops::Range;
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex, MutexGuard};

use clawhdf5_format::error::FormatError;
use clawhdf5_format::storage::Storage;

/// Default block size: 1 MiB, the size `docs/design/range-reads.md` §2
/// measured (metadata of the test files in 1–15 blocks; the chunk index of a
/// netCDF file spread over 14 blocks of a 48 MB file).
pub const DEFAULT_BLOCK_SIZE: u64 = 1 << 20;

/// Settings of a [`BlockCache`].
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CacheConfig {
    /// Size of a block, in bytes; every fetch is whole, aligned blocks
    /// (the last block of the file is shorter).
    pub block_size: u64,
    /// Byte budget of cached blocks.
    pub capacity: u64,
    /// Two runs of missing blocks separated by at most this many bytes of
    /// uncached blocks are fetched as one request (the gap included).
    pub coalesce_gap: u64,
    /// Largest single request, in bytes (rounded down to whole blocks, at
    /// least one block). Longer runs are split, so a backend can fetch the
    /// pieces in parallel.
    pub max_request: u64,
}

impl Default for CacheConfig {
    fn default() -> Self {
        CacheConfig {
            block_size: DEFAULT_BLOCK_SIZE,
            capacity: 64 << 20,
            coalesce_gap: DEFAULT_BLOCK_SIZE,
            max_request: 8 << 20,
        }
    }
}

/// What a [`BlockCache`] has done so far.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct CacheStats {
    /// Reads served (`read_at` calls plus ranges of `read_ranges` calls).
    pub reads: u64,
    /// Block lookups that found the block cached.
    pub hits: u64,
    /// Block lookups that had to fetch the block (or wait for it).
    pub misses: u64,
    /// Block lookups that waited for another reader's fetch of the block.
    pub waits: u64,
    /// Ranges requested from the backend: for HTTP, one request each.
    pub requests: u64,
    /// `read_ranges` calls made on the backend.
    pub fetch_calls: u64,
    /// Bytes fetched from the backend.
    pub bytes_fetched: u64,
    /// Blocks evicted to stay within the budget.
    pub evictions: u64,
    /// Bytes currently cached.
    pub cached_bytes: u64,
}

type Block = Arc<[u8]>;

/// A fetch in progress: the readers waiting for a block wait on this.
struct Flight {
    result: Mutex<Option<Result<Block, FormatError>>>,
    done: Condvar,
}

impl Flight {
    fn new() -> Arc<Flight> {
        Arc::new(Flight {
            result: Mutex::new(None),
            done: Condvar::new(),
        })
    }

    fn finish(&self, r: Result<Block, FormatError>) {
        let mut slot = lock(&self.result);
        if slot.is_none() {
            *slot = Some(r);
        }
        self.done.notify_all();
    }

    fn wait(&self) -> Result<Block, FormatError> {
        let mut slot = lock(&self.result);
        loop {
            if let Some(r) = slot.as_ref() {
                return r.clone();
            }
            slot = self
                .done
                .wait(slot)
                .unwrap_or_else(std::sync::PoisonError::into_inner);
        }
    }
}

enum Slot {
    Ready { data: Block, tick: u64 },
    Pending(Arc<Flight>),
}

#[derive(Default)]
struct State {
    blocks: HashMap<u64, Slot>,
    /// tick -> block index, oldest first.
    lru: BTreeMap<u64, u64>,
    tick: u64,
    bytes: u64,
}

fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
    m.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
}

#[derive(Default)]
struct Counters {
    reads: AtomicU64,
    hits: AtomicU64,
    misses: AtomicU64,
    waits: AtomicU64,
    requests: AtomicU64,
    fetch_calls: AtomicU64,
    bytes_fetched: AtomicU64,
    evictions: AtomicU64,
}

/// A [`Storage`] that serves reads of `inner` from an LRU cache of
/// fixed-size blocks. See the [module documentation](self).
pub struct BlockCache<S> {
    inner: S,
    config: CacheConfig,
    len: u64,
    state: Mutex<State>,
    counters: Counters,
}

/// Fails every flight a fetch claimed and did not complete (an error or a
/// panic in the backend), so no reader waits forever. The waiting readers
/// get the fetch's error.
struct FlightGuard<'a, S> {
    cache: &'a BlockCache<S>,
    flights: Vec<(u64, Arc<Flight>)>,
    error: Option<FormatError>,
}

impl<S> FlightGuard<'_, S> {
    /// Record `e` for the readers waiting on this fetch; returns it.
    fn fail(&mut self, e: FormatError) -> FormatError {
        self.error = Some(e.clone());
        e
    }
}

impl<S> Drop for FlightGuard<'_, S> {
    fn drop(&mut self) {
        if self.flights.is_empty() {
            return;
        }
        let mut st = lock(&self.cache.state);
        for (i, f) in &self.flights {
            if matches!(st.blocks.get(i), Some(Slot::Pending(p)) if Arc::ptr_eq(p, f)) {
                st.blocks.remove(i);
            }
        }
        drop(st);
        let error = self
            .error
            .take()
            .unwrap_or_else(|| FormatError::Storage("the fetch of this block failed".into()));
        for (_, f) in &self.flights {
            f.finish(Err(error.clone()));
        }
    }
}

impl<S: Storage> BlockCache<S> {
    /// Cache `inner` with the given settings. The length of the file is
    /// taken from `inner` once, here: a remote file is pinned at open.
    pub fn new(inner: S, mut config: CacheConfig) -> Self {
        config.block_size = config.block_size.max(512);
        config.capacity = config.capacity.max(config.block_size);
        config.max_request = (config.max_request / config.block_size).max(1) * config.block_size;
        let len = inner.len();
        BlockCache {
            inner,
            config,
            len,
            state: Mutex::new(State::default()),
            counters: Counters::default(),
        }
    }

    /// The backend.
    pub fn inner(&self) -> &S {
        &self.inner
    }

    /// The settings in use (after rounding).
    pub fn config(&self) -> &CacheConfig {
        &self.config
    }

    /// Counters since the cache was made (or last [`reset_stats`](Self::reset_stats)).
    pub fn stats(&self) -> CacheStats {
        let c = &self.counters;
        CacheStats {
            reads: c.reads.load(Ordering::Relaxed),
            hits: c.hits.load(Ordering::Relaxed),
            misses: c.misses.load(Ordering::Relaxed),
            waits: c.waits.load(Ordering::Relaxed),
            requests: c.requests.load(Ordering::Relaxed),
            fetch_calls: c.fetch_calls.load(Ordering::Relaxed),
            bytes_fetched: c.bytes_fetched.load(Ordering::Relaxed),
            evictions: c.evictions.load(Ordering::Relaxed),
            cached_bytes: lock(&self.state).bytes,
        }
    }

    /// Zero the counters (the cached blocks stay).
    pub fn reset_stats(&self) {
        let c = &self.counters;
        for a in [
            &c.reads,
            &c.hits,
            &c.misses,
            &c.waits,
            &c.requests,
            &c.fetch_calls,
            &c.bytes_fetched,
            &c.evictions,
        ] {
            a.store(0, Ordering::Relaxed);
        }
    }

    /// Drop every cached block.
    pub fn clear(&self) {
        let mut st = lock(&self.state);
        st.blocks.retain(|_, s| matches!(s, Slot::Pending(_)));
        st.lru.clear();
        st.bytes = 0;
    }

    /// Fetch the blocks covering `[offset, offset + len)` now (readahead),
    /// keeping them cached.
    pub fn prefetch(&self, offset: u64, len: u64) -> Result<(), FormatError> {
        if self.inner.as_contiguous().is_some() {
            return Ok(());
        }
        let Some(blocks) = self.block_span(offset, len) else {
            return Ok(());
        };
        self.blocks(&blocks.collect::<Vec<_>>(), true).map(|_| ())
    }

    /// Put bytes already fetched (such as the first block, which a backend
    /// may get while it probes the file's length) into the cache: `bytes`
    /// are the file's bytes at `offset`. Only whole blocks (or the file's
    /// last, short block) are kept; a block already cached is left alone.
    /// The bytes count as one request in [`CacheStats`].
    pub fn insert(&self, offset: u64, bytes: &[u8]) {
        self.counters.requests.fetch_add(1, Ordering::Relaxed);
        self.counters
            .bytes_fetched
            .fetch_add(bytes.len() as u64, Ordering::Relaxed);
        let bs = self.config.block_size;
        let end = offset.saturating_add(bytes.len() as u64).min(self.len);
        let mut i = offset.div_ceil(bs);
        let mut st = lock(&self.state);
        while i * bs < end {
            let start = i * bs;
            let block_end = (start + bs).min(self.len);
            if block_end > end {
                break;
            }
            if !st.blocks.contains_key(&i) {
                let rel = (start - offset) as usize..(block_end - offset) as usize;
                self.keep(&mut st, i, Arc::from(&bytes[rel]));
            }
            i += 1;
        }
        self.evict(&mut st);
    }

    /// Block indices covering `[offset, offset + len)`, clamped to the file.
    fn block_span(&self, offset: u64, len: u64) -> Option<Range<u64>> {
        let end = offset.saturating_add(len).min(self.len);
        if offset >= end {
            return None;
        }
        let bs = self.config.block_size;
        Some(offset / bs..(end - 1) / bs + 1)
    }

    fn block_len(&self, i: u64) -> u64 {
        let start = i * self.config.block_size;
        (start + self.config.block_size).min(self.len) - start
    }

    fn keep(&self, st: &mut State, i: u64, data: Block) {
        st.tick += 1;
        let tick = st.tick;
        st.bytes += <[u8]>::len(&data) as u64;
        st.lru.insert(tick, i);
        st.blocks.insert(i, Slot::Ready { data, tick });
    }

    fn evict(&self, st: &mut State) {
        while st.bytes > self.config.capacity {
            let Some((_, i)) = st.lru.pop_first() else {
                break;
            };
            if let Some(Slot::Ready { data, .. }) = st.blocks.remove(&i) {
                st.bytes -= <[u8]>::len(&data) as u64;
                self.counters.evictions.fetch_add(1, Ordering::Relaxed);
            }
        }
    }

    /// The blocks `wanted` (sorted, distinct), fetching the missing ones.
    fn blocks(&self, wanted: &[u64], may_keep: bool) -> Result<HashMap<u64, Block>, FormatError> {
        let mut have = HashMap::with_capacity(wanted.len());
        let mut waits = Vec::new();
        let mut guard = FlightGuard {
            cache: self,
            flights: Vec::new(),
            error: None,
        };
        {
            let mut st = lock(&self.state);
            for &i in wanted {
                match st.blocks.get(&i) {
                    Some(Slot::Ready { data, tick }) => {
                        let (data, old) = (data.clone(), *tick);
                        st.tick += 1;
                        let tick = st.tick;
                        st.lru.remove(&old);
                        st.lru.insert(tick, i);
                        if let Some(Slot::Ready { tick: t, .. }) = st.blocks.get_mut(&i) {
                            *t = tick;
                        }
                        self.counters.hits.fetch_add(1, Ordering::Relaxed);
                        have.insert(i, data);
                    }
                    Some(Slot::Pending(f)) => {
                        self.counters.misses.fetch_add(1, Ordering::Relaxed);
                        self.counters.waits.fetch_add(1, Ordering::Relaxed);
                        waits.push((i, f.clone()));
                    }
                    None => {
                        self.counters.misses.fetch_add(1, Ordering::Relaxed);
                        let f = Flight::new();
                        st.blocks.insert(i, Slot::Pending(f.clone()));
                        guard.flights.push((i, f));
                    }
                }
            }
            // Fill small gaps between runs with blocks nobody has, so the
            // runs merge into one request.
            if self.config.coalesce_gap > 0 && guard.flights.len() > 1 {
                let gap_blocks = self.config.coalesce_gap / self.config.block_size;
                let mut extra = Vec::new();
                for w in guard.flights.windows(2) {
                    let (a, b) = (w[0].0, w[1].0);
                    let gap = b - a - 1;
                    if gap == 0 || gap > gap_blocks {
                        continue;
                    }
                    if (a + 1..b).all(|j| !st.blocks.contains_key(&j)) {
                        extra.extend(a + 1..b);
                    }
                }
                for j in extra {
                    let f = Flight::new();
                    st.blocks.insert(j, Slot::Pending(f.clone()));
                    guard.flights.push((j, f));
                }
                guard.flights.sort_by_key(|(i, _)| *i);
            }
        }

        if !guard.flights.is_empty() {
            let runs = self.runs(&guard.flights);
            let missing_bytes: u64 = runs.iter().map(|r| r.end - r.start).sum();
            let keep = may_keep && missing_bytes <= self.config.capacity / 2;
            self.counters.fetch_calls.fetch_add(1, Ordering::Relaxed);
            self.counters
                .requests
                .fetch_add(runs.len() as u64, Ordering::Relaxed);
            let fetched = match self.inner.read_ranges(&runs) {
                Ok(f) => f,
                Err(e) => return Err(guard.fail(e)),
            };
            if fetched.len() != runs.len() {
                return Err(guard.fail(FormatError::Storage(format!(
                    "backend returned {} ranges for {} requested",
                    fetched.len(),
                    runs.len()
                ))));
            }
            let mut got: Vec<(u64, Block)> = Vec::with_capacity(guard.flights.len());
            for (run, bytes) in runs.iter().zip(&fetched) {
                self.counters
                    .bytes_fetched
                    .fetch_add(bytes.len() as u64, Ordering::Relaxed);
                if bytes.len() as u64 != run.end - run.start {
                    return Err(guard.fail(FormatError::Storage(format!(
                        "short read from the backend: {} of {} bytes at offset {}",
                        bytes.len(),
                        run.end - run.start,
                        run.start
                    ))));
                }
                let bs = self.config.block_size;
                let mut i = run.start / bs;
                let mut pos = 0usize;
                while pos < bytes.len() {
                    let n = self.block_len(i) as usize;
                    got.push((i, Arc::from(&bytes[pos..pos + n])));
                    pos += n;
                    i += 1;
                }
            }
            drop(fetched);
            let flights = std::mem::take(&mut guard.flights);
            let mut st = lock(&self.state);
            let mut by_index: HashMap<u64, Block> = got.into_iter().collect();
            for (i, f) in &flights {
                let Some(data) = by_index.remove(i) else {
                    continue;
                };
                let ours = matches!(st.blocks.get(i), Some(Slot::Pending(p)) if Arc::ptr_eq(p, f));
                if ours {
                    if keep {
                        self.keep(&mut st, *i, data.clone());
                    } else {
                        st.blocks.remove(i);
                    }
                }
                f.finish(Ok(data.clone()));
                have.insert(*i, data);
            }
            self.evict(&mut st);
            drop(st);
            // Any flight without data (cannot happen: every run is split
            // into its blocks) is failed rather than left waiting.
            guard.flights = flights
                .into_iter()
                .filter(|(i, _)| !have.contains_key(i))
                .collect();
            if !guard.flights.is_empty() {
                return Err(FormatError::Storage(
                    "backend did not return every block".into(),
                ));
            }
        }

        for (i, f) in waits {
            have.insert(i, f.wait()?);
        }
        Ok(have)
    }

    /// Byte ranges of the runs of consecutive blocks in `flights` (sorted),
    /// each at most `max_request` long.
    fn runs(&self, flights: &[(u64, Arc<Flight>)]) -> Vec<Range<u64>> {
        let bs = self.config.block_size;
        let per_request = self.config.max_request / bs;
        let mut runs: Vec<(u64, u64)> = Vec::new();
        for &(i, _) in flights {
            match runs.last_mut() {
                Some((first, last)) if *last + 1 == i && i - *first < per_request => *last = i,
                _ => runs.push((i, i)),
            }
        }
        runs.into_iter()
            .map(|(a, b)| a * bs..(b * bs + self.block_len(b)))
            .collect()
    }

    /// Copy `[offset, end)` out of `blocks`.
    fn assemble(&self, offset: u64, end: u64, blocks: &HashMap<u64, Block>) -> Vec<u8> {
        let bs = self.config.block_size;
        let mut out = Vec::with_capacity((end - offset) as usize);
        let mut pos = offset;
        while pos < end {
            let i = pos / bs;
            let block = &blocks[&i];
            let from = (pos - i * bs) as usize;
            let to = ((end - i * bs) as usize).min(<[u8]>::len(block));
            out.extend_from_slice(&block[from..to]);
            pos = i * bs + to as u64;
        }
        out
    }
}

impl<S: Storage> Storage for BlockCache<S> {
    fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, FormatError> {
        if let Some(all) = self.inner.as_contiguous() {
            return all.read_at(offset, len);
        }
        self.counters.reads.fetch_add(1, Ordering::Relaxed);
        let Some(span) = self.block_span(offset, len as u64) else {
            return Ok(Cow::Owned(Vec::new()));
        };
        let end = offset.saturating_add(len as u64).min(self.len);
        let blocks = self.blocks(&span.collect::<Vec<_>>(), true)?;
        Ok(Cow::Owned(self.assemble(offset, end, &blocks)))
    }

    fn len(&self) -> u64 {
        self.len
    }

    fn read_ranges(&self, ranges: &[Range<u64>]) -> Result<Vec<Cow<'_, [u8]>>, FormatError> {
        if let Some(all) = self.inner.as_contiguous() {
            return all.read_ranges(ranges);
        }
        self.counters
            .reads
            .fetch_add(ranges.len() as u64, Ordering::Relaxed);
        let mut wanted = BTreeSet::new();
        for r in ranges {
            if r.end < r.start {
                return Err(FormatError::Storage(
                    "read range ends before it starts".into(),
                ));
            }
            if let Some(span) = self.block_span(r.start, r.end - r.start) {
                wanted.extend(span);
            }
        }
        let wanted: Vec<u64> = wanted.into_iter().collect();
        let blocks = self.blocks(&wanted, true)?;
        Ok(ranges
            .iter()
            .map(|r| {
                let end = r.end.min(self.len);
                if r.start >= end {
                    Cow::Owned(Vec::new())
                } else {
                    Cow::Owned(self.assemble(r.start, end, &blocks))
                }
            })
            .collect())
    }

    fn as_contiguous(&self) -> Option<&[u8]> {
        self.inner.as_contiguous()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use clawhdf5_format::storage::CountingStorage;

    fn data(n: usize) -> Vec<u8> {
        (0..n).map(|i| (i * 7 + i / 251) as u8).collect()
    }

    fn cache(n: usize, config: CacheConfig) -> (Vec<u8>, BlockCache<CountingStorage>) {
        let d = data(n);
        (d.clone(), BlockCache::new(CountingStorage::new(d), config))
    }

    fn small() -> CacheConfig {
        CacheConfig {
            block_size: 1024,
            capacity: 8 * 1024,
            coalesce_gap: 0,
            max_request: 4 * 1024,
        }
    }

    #[test]
    fn reads_match_the_data_and_hit_the_cache() {
        let mut cfg = small();
        cfg.capacity = 64 * 1024;
        let (d, c) = cache(10_000, cfg);
        for (off, len) in [
            (0, 10),
            (1000, 100),
            (1020, 10),
            (9990, 100),
            (20_000, 5),
            (0, 10_000),
        ] {
            let got = c.read_at(off, len).unwrap();
            let s = (off as usize).min(d.len());
            let e = (off as usize + len).min(d.len());
            assert_eq!(&*got, &d[s..e], "{off} {len}");
        }
        let before = c.inner().reads();
        assert_eq!(&*c.read_at(5000, 3000).unwrap(), &d[5000..8000]);
        assert_eq!(c.inner().reads(), before, "all cached");
        assert!(c.stats().hits > 0);
    }

    #[test]
    fn read_ranges_coalesces_consecutive_missing_blocks() {
        let (d, c) = cache(20_000, small());
        let ranges = [100..200, 1500..1600, 2100..3000, 9000..9100];
        let got = c.read_ranges(&ranges).unwrap();
        for (r, g) in ranges.iter().zip(&got) {
            assert_eq!(&**g, &d[r.start as usize..r.end as usize]);
        }
        // Blocks 0-2 are one run, block 8 another: one backend call, two ranges.
        assert_eq!(c.stats().fetch_calls, 1);
        assert_eq!(c.stats().requests, 2);
        assert_eq!(c.inner().reads(), 2);
    }

    #[test]
    fn a_small_gap_is_fetched_to_merge_runs() {
        let mut cfg = small();
        cfg.coalesce_gap = 1024;
        let (d, c) = cache(20_000, cfg);
        let ranges = [0..10, 2048..2058, 5000..5010];
        let got = c.read_ranges(&ranges).unwrap();
        for (r, g) in ranges.iter().zip(&got) {
            assert_eq!(&**g, &d[r.start as usize..r.end as usize]);
        }
        // Blocks 0 and 2 merge over block 1; block 4 is two blocks away.
        assert_eq!(c.stats().requests, 2);
    }

    #[test]
    fn long_runs_are_split_at_max_request() {
        let (d, c) = cache(20_000, small());
        assert_eq!(&*c.read_at(0, 10_000).unwrap(), &d[..10_000]);
        // 10 blocks, 4 per request: 3 requests in one call.
        assert_eq!(c.stats().requests, 3);
        assert_eq!(c.stats().fetch_calls, 1);
    }

    #[test]
    fn lru_stays_within_budget_and_evicts_oldest() {
        let (d, c) = cache(40_000, small());
        for i in 0..8u64 {
            c.read_at(i * 1024, 1).unwrap();
        }
        c.read_at(0, 1).unwrap(); // block 0 is now the newest
        c.read_at(8 * 1024, 1).unwrap(); // evicts block 1
        let s = c.stats();
        assert!(s.cached_bytes <= 8 * 1024);
        assert_eq!(s.evictions, 1);
        let before = c.inner().reads();
        c.read_at(0, 1).unwrap();
        assert_eq!(c.inner().reads(), before, "block 0 kept");
        assert_eq!(&*c.read_at(1024, 5).unwrap(), &d[1024..1029]);
        assert_eq!(c.inner().reads(), before + 1, "block 1 refetched");
    }

    #[test]
    fn large_reads_do_not_flush_the_cache() {
        let (d, c) = cache(40_000, small());
        c.read_at(0, 1).unwrap();
        assert_eq!(&*c.read_at(10_000, 20_000).unwrap(), &d[10_000..30_000]);
        assert_eq!(c.stats().cached_bytes, 1024, "only block 0 kept");
    }

    #[test]
    fn insert_keeps_whole_blocks_only() {
        let (d, c) = cache(5000, small());
        c.insert(0, &d[..2500]);
        assert_eq!(c.stats().cached_bytes, 2048);
        c.insert(4096, &d[4096..]);
        assert_eq!(c.stats().cached_bytes, 2048 + 904);
        assert_eq!(&*c.read_at(0, 2048).unwrap(), &d[..2048]);
        assert_eq!(&*c.read_at(4500, 500).unwrap(), &d[4500..]);
        assert_eq!(c.inner().reads(), 0);
    }

    /// A backend that fails its first `n` fetches.
    struct Flaky {
        data: Vec<u8>,
        fail: AtomicU64,
    }

    impl Storage for Flaky {
        fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, FormatError> {
            if self
                .fail
                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |n| n.checked_sub(1))
                .is_ok()
            {
                return Err(FormatError::Storage("boom".into()));
            }
            self.data
                .as_slice()
                .read_at(offset, len)
                .map(|c| Cow::Owned(c.into_owned()))
        }
        fn len(&self) -> u64 {
            self.data.len() as u64
        }
    }

    #[test]
    fn a_failed_fetch_is_an_error_and_is_not_cached() {
        let d = data(5000);
        let c = BlockCache::new(
            Flaky {
                data: d.clone(),
                fail: AtomicU64::new(1),
            },
            small(),
        );
        assert!(c.read_at(0, 10).is_err());
        assert_eq!(&*c.read_at(0, 10).unwrap(), &d[..10]);
    }

    /// A backend that returns fewer bytes than asked inside the file.
    struct Short(Vec<u8>);

    impl Storage for Short {
        fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, FormatError> {
            let got = self.0.as_slice().read_at(offset, len)?;
            Ok(Cow::Owned(got[..got.len() / 2].to_vec()))
        }
        fn len(&self) -> u64 {
            self.0.len() as u64
        }
    }

    #[test]
    fn a_short_backend_read_is_an_error() {
        let c = BlockCache::new(Short(data(5000)), small());
        assert!(c.read_at(0, 10).is_err());
        assert!(c.read_at(0, 10).is_err(), "not cached");
    }

    #[test]
    fn contiguous_backends_pass_through() {
        let d = data(5000);
        let c = BlockCache::new(d.clone(), small());
        assert_eq!(c.as_contiguous(), Some(&d[..]));
        assert!(matches!(c.read_at(0, 10).unwrap(), Cow::Borrowed(_)));
        assert_eq!(c.stats().requests, 0);
    }

    /// A slow backend: concurrent readers of the same blocks share fetches.
    struct Slow {
        data: Vec<u8>,
        fetched: Mutex<Vec<Range<u64>>>,
    }

    impl Storage for Slow {
        fn read_at(&self, offset: u64, len: usize) -> Result<Cow<'_, [u8]>, FormatError> {
            std::thread::sleep(std::time::Duration::from_millis(30));
            lock(&self.fetched).push(offset..offset + len as u64);
            self.data
                .as_slice()
                .read_at(offset, len)
                .map(|c| Cow::Owned(c.into_owned()))
        }
        fn len(&self) -> u64 {
            self.data.len() as u64
        }
    }

    /// A slow backend whose fetches all fail with a telling error.
    struct SlowFailing;

    impl Storage for SlowFailing {
        fn read_at(&self, _: u64, _: usize) -> Result<Cow<'_, [u8]>, FormatError> {
            std::thread::sleep(std::time::Duration::from_millis(50));
            Err(FormatError::Storage("the file changed while open".into()))
        }
        fn len(&self) -> u64 {
            1 << 20
        }
    }

    #[test]
    fn readers_waiting_on_a_failed_fetch_get_its_error() {
        let c = BlockCache::new(SlowFailing, small());
        std::thread::scope(|s| {
            let hs: Vec<_> = (0..4)
                .map(|_| s.spawn(|| c.read_at(0, 10).map(|b| b.len())))
                .collect();
            for h in hs {
                let e = h.join().unwrap().unwrap_err().to_string();
                assert!(e.contains("changed while open"), "{e}");
            }
        });
        assert!(c.stats().waits > 0);
    }

    #[test]
    fn concurrent_readers_never_fetch_a_block_twice() {
        let d = data(64 * 1024);
        let cfg = CacheConfig {
            block_size: 1024,
            capacity: 1 << 20,
            coalesce_gap: 0,
            max_request: 1024,
        };
        let c = BlockCache::new(
            Slow {
                data: d.clone(),
                fetched: Mutex::new(Vec::new()),
            },
            cfg,
        );
        std::thread::scope(|s| {
            for t in 0..8u64 {
                let (c, d) = (&c, &d);
                s.spawn(move || {
                    for k in 0..16u64 {
                        let off = ((k * 3 + t) % 32) * 1024 + 100;
                        let got = c.read_at(off, 2000).unwrap();
                        assert_eq!(&*got, &d[off as usize..off as usize + 2000]);
                    }
                });
            }
        });
        let fetched = lock(&c.inner().fetched).clone();
        let mut blocks: Vec<u64> = fetched.iter().map(|r| r.start / 1024).collect();
        let n = blocks.len();
        blocks.sort_unstable();
        blocks.dedup();
        assert_eq!(n, blocks.len(), "a block was fetched twice: {fetched:?}");
        assert!(c.stats().waits > 0, "readers should have shared fetches");
    }
}