irontide_engine/

disk_backend.rs

#![allow(
    clippy::cast_possible_truncation,
    clippy::cast_precision_loss,
    clippy::cast_possible_wrap,
    clippy::cast_sign_loss,
    reason = "M175: disk I/O sizes bounded by piece_length (u32 by construction in Lengths::new); offset arithmetic stays within file lengths"
)]

//! Pluggable disk I/O backend abstraction.
//!
//! The [`DiskIoBackend`] trait defines the interface for session-level disk I/O,
//! allowing custom storage backends (POSIX, disabled/null, etc.).
//! [`DisabledDiskIo`] is a no-op backend useful for network throughput benchmarking.

use std::collections::HashMap;
use std::sync::Arc;

use parking_lot::{Mutex, RwLock};

use bytes::Bytes;
use irontide_core::{Id20, Id32};
use irontide_storage::TorrentStorage;

use crate::buffer_pool::BufferPool;
use crate::disk::DiskConfig;

/// Aggregate I/O statistics for a disk backend.
#[derive(Debug, Clone, Default)]
pub struct DiskIoStats {
    /// Total bytes read from storage.
    pub read_bytes: u64,
    /// Total bytes written to storage.
    pub write_bytes: u64,
    /// Number of read requests satisfied from cache.
    pub cache_hits: u64,
    /// Number of read requests that required disk access.
    pub cache_misses: u64,
    /// Current size of the write buffer in bytes.
    pub write_buffer_bytes: usize,
    /// Current size of the read cache in bytes (M102).
    pub read_cache_bytes: usize,
    /// Total number of entries in the buffer pool (M102).
    pub pool_entries: usize,
    /// Number of prefetch insertions into the read cache (M102).
    pub prefetch_count: u64,
    /// Number of ARC evictions from the read cache (M102).
    pub eviction_count: u64,
    /// Number of Writing-to-Skeleton demotions (M102).
    pub skeleton_count: u64,
}

/// Trait for pluggable disk I/O backends.
///
/// Implementations are shared across the session via `Arc<dyn DiskIoBackend>`.
/// All methods must be safe to call from multiple threads concurrently.
pub trait DiskIoBackend: Send + Sync {
    /// Human-readable backend name (e.g., "posix", "disabled").
    fn name(&self) -> &str;

    /// Register a torrent's storage for I/O operations.
    fn register(&self, info_hash: Id20, storage: Arc<dyn TorrentStorage>);

    /// Unregister a torrent's storage.
    fn unregister(&self, info_hash: Id20);

    /// Write a chunk of piece data. If `flush` is true, persist immediately.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage write fails.
    fn write_chunk(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        data: Bytes,
        flush: bool,
    ) -> crate::Result<()>;

    /// Read a chunk of piece data. `volatile` hints the data won't be re-read soon.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage read fails.
    fn read_chunk(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        length: u32,
        volatile: bool,
    ) -> crate::Result<Bytes>;

    /// Read an entire piece from storage, returning its raw bytes.
    ///
    /// For backends with a write buffer, the piece is flushed before reading.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage read fails.
    fn read_piece(&self, info_hash: Id20, piece: u32) -> crate::Result<Vec<u8>>;

    /// Verify a piece against its expected SHA-1 hash.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage read fails.
    fn hash_piece(&self, info_hash: Id20, piece: u32, expected: &Id20) -> crate::Result<bool>;

    /// Verify a piece against its expected SHA-256 hash (BEP 52).
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage read fails.
    fn hash_piece_v2(&self, info_hash: Id20, piece: u32, expected: &Id32) -> crate::Result<bool>;

    /// Compute SHA-256 hash of a single block within a piece (BEP 52 Merkle).
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage read fails.
    fn hash_block(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        length: u32,
    ) -> crate::Result<Id32>;

    /// Discard buffered data for a piece (e.g., after hash failure).
    fn clear_piece(&self, info_hash: Id20, piece: u32);

    /// Flush buffered writes for a piece to persistent storage.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage flush fails.
    fn flush_piece(&self, info_hash: Id20, piece: u32) -> crate::Result<()>;

    /// Flush all buffered writes across all torrents.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage flush fails.
    fn flush_all(&self) -> crate::Result<()>;

    /// Return piece indices currently held in cache (for `SuggestPiece`, M44).
    fn cached_pieces(&self, info_hash: Id20) -> Vec<u32>;

    /// Return current I/O statistics.
    fn stats(&self) -> DiskIoStats;

    /// Write a block directly to storage from two slices (vectored write).
    ///
    /// Bypasses the buffer pool entirely — data goes straight to the
    /// underlying `TorrentStorage`. The two slices represent contiguous
    /// block data split across a ring-buffer wrap boundary.
    ///
    /// This is the zero-copy direct-pwrite path used by peer tasks.
    ///
    /// # Errors
    ///
    /// Returns an error if the underlying storage write fails.
    fn write_block_direct(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        s0: &[u8],
        s1: &[u8],
    ) -> crate::Result<()>;
}

/// No-op disk I/O backend for network throughput benchmarking.
///
/// All writes succeed silently, reads return zeroed bytes, and hash
/// verifications always pass.
pub struct DisabledDiskIo;

impl DiskIoBackend for DisabledDiskIo {
    fn name(&self) -> &'static str {
        "disabled"
    }

    fn register(&self, _info_hash: Id20, _storage: Arc<dyn TorrentStorage>) {}

    fn unregister(&self, _info_hash: Id20) {}

    fn write_chunk(
        &self,
        _info_hash: Id20,
        _piece: u32,
        _begin: u32,
        _data: Bytes,
        _flush: bool,
    ) -> crate::Result<()> {
        Ok(())
    }

    fn read_chunk(
        &self,
        _info_hash: Id20,
        _piece: u32,
        _begin: u32,
        length: u32,
        _volatile: bool,
    ) -> crate::Result<Bytes> {
        Ok(Bytes::from(vec![0u8; length as usize]))
    }

    fn read_piece(&self, _info_hash: Id20, _piece: u32) -> crate::Result<Vec<u8>> {
        Ok(Vec::new())
    }

    fn hash_piece(&self, _info_hash: Id20, _piece: u32, _expected: &Id20) -> crate::Result<bool> {
        Ok(true)
    }

    fn hash_piece_v2(
        &self,
        _info_hash: Id20,
        _piece: u32,
        _expected: &Id32,
    ) -> crate::Result<bool> {
        Ok(true)
    }

    fn hash_block(
        &self,
        _info_hash: Id20,
        _piece: u32,
        _begin: u32,
        _length: u32,
    ) -> crate::Result<Id32> {
        Ok(Id32([0u8; 32]))
    }

    fn clear_piece(&self, _info_hash: Id20, _piece: u32) {}

    fn flush_piece(&self, _info_hash: Id20, _piece: u32) -> crate::Result<()> {
        Ok(())
    }

    fn flush_all(&self) -> crate::Result<()> {
        Ok(())
    }

    fn cached_pieces(&self, _info_hash: Id20) -> Vec<u32> {
        vec![]
    }

    fn stats(&self) -> DiskIoStats {
        DiskIoStats::default()
    }

    fn write_block_direct(
        &self,
        _info_hash: Id20,
        _piece: u32,
        _begin: u32,
        _s0: &[u8],
        _s1: &[u8],
    ) -> crate::Result<()> {
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// PosixDiskIo — unified BufferPool backend
// ---------------------------------------------------------------------------

/// POSIX disk I/O backend with unified buffer pool (combined read cache +
/// write buffer under a single byte budget).
///
/// Replaces the previous separate `ArcCache` + `WriteBuffer` design with
/// a single [`BufferPool`] that uses a byte-budget ARC for read caching and
/// tracks in-flight writes as `Writing` entries.
pub struct PosixDiskIo {
    storages: RwLock<HashMap<Id20, Arc<dyn TorrentStorage>>>,
    pool: Mutex<BufferPool>,
    stats: Mutex<DiskIoStats>,
}

impl PosixDiskIo {
    /// Create a new POSIX disk I/O backend with the given configuration.
    #[must_use]
    pub fn new(config: &DiskConfig) -> Self {
        let mut pool = BufferPool::with_capacity(config.buffer_pool_capacity);
        pool.set_mlock(config.enable_mlock);
        Self {
            storages: RwLock::new(HashMap::new()),
            pool: Mutex::new(pool),
            stats: Mutex::new(DiskIoStats::default()),
        }
    }

    fn get_storage(&self, info_hash: Id20) -> crate::Result<Arc<dyn TorrentStorage>> {
        self.storages
            .read()
            .get(&info_hash)
            .cloned()
            .ok_or_else(|| {
                crate::Error::Storage(irontide_storage::Error::Io(std::io::Error::new(
                    std::io::ErrorKind::NotFound,
                    "torrent not registered",
                )))
            })
    }
}

impl DiskIoBackend for PosixDiskIo {
    fn name(&self) -> &'static str {
        "posix"
    }

    fn register(&self, info_hash: Id20, storage: Arc<dyn TorrentStorage>) {
        self.storages.write().insert(info_hash, storage);
    }

    fn unregister(&self, info_hash: Id20) {
        self.storages.write().remove(&info_hash);
        self.pool.lock().clear_torrent(info_hash);
    }

    fn write_chunk(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        data: Bytes,
        flush: bool,
    ) -> crate::Result<()> {
        let len = data.len();

        if flush {
            let storage = self.get_storage(info_hash)?;
            storage.write_chunk(piece, begin, &data)?;
            self.stats.lock().write_bytes += len as u64;
            return Ok(());
        }

        // Buffer the write in the pool. Pass piece_size=0 since we don't know
        // it here — completion detection is handled by ChunkTracker in
        // torrent.rs, not by the buffer pool.
        let key = (info_hash, piece);
        self.pool.lock().write_block(key, begin, data, 0);
        self.stats.lock().write_bytes += len as u64;
        Ok(())
    }

    fn read_chunk(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        length: u32,
        volatile: bool,
    ) -> crate::Result<Bytes> {
        let key = (info_hash, piece);

        // 1. Check buffer pool (Writing entries + Cached entries)
        if !volatile {
            let mut pool = self.pool.lock();
            if let Some(data) = pool.read_block(key, begin, length as usize) {
                drop(pool);
                self.stats.lock().cache_hits += 1;
                return Ok(data);
            }
        }
        self.stats.lock().cache_misses += 1;

        // 2. Read full piece from storage and cache it (unless volatile)
        let storage = self.get_storage(info_hash)?;
        if !volatile {
            if let Ok(piece_data) = storage.read_piece(piece) {
                let piece_bytes = Bytes::from(piece_data);
                self.stats.lock().read_bytes += piece_bytes.len() as u64;

                let mut pool = self.pool.lock();
                pool.prefetch_piece(key, piece_bytes.clone());
                drop(pool);

                // Return the requested slice
                let end = (begin as usize).saturating_add(length as usize);
                if end <= piece_bytes.len() {
                    return Ok(piece_bytes.slice(begin as usize..end));
                }
                // Fall through to chunk read if slice is out of bounds
                // (shouldn't happen in practice, but defensive)
            } else {
                // Fall through to chunk read on piece-level failure
            }
        }

        // 3. Volatile path or fallback: read just the requested chunk
        let data = storage.read_chunk(piece, begin, length)?;
        let bytes = Bytes::from(data);
        self.stats.lock().read_bytes += bytes.len() as u64;
        Ok(bytes)
    }

    fn read_piece(&self, info_hash: Id20, piece: u32) -> crate::Result<Vec<u8>> {
        self.flush_piece(info_hash, piece)?;
        let storage = self.get_storage(info_hash)?;
        let data = storage.read_piece(piece)?;
        self.stats.lock().read_bytes += data.len() as u64;
        Ok(data)
    }

    fn hash_piece(&self, info_hash: Id20, piece: u32, expected: &Id20) -> crate::Result<bool> {
        let key = (info_hash, piece);

        // Try hash from cache: take all blocks from the Writing entry and hash
        // them in memory, avoiding a disk round-trip. This works because
        // hash_piece is only called after ChunkTracker confirms the piece is
        // complete, so all blocks should be present.
        let cached_data = {
            let mut pool = self.pool.lock();
            pool.take_all_blocks(key)
        };

        if let Some(data) = cached_data {
            let hash = irontide_core::sha1(&data);
            if hash == *expected {
                // Hash pass: flush to disk + promote to read cache.
                let storage = self.get_storage(info_hash)?;
                storage.write_chunk(piece, 0, &data)?;
                self.stats.lock().write_bytes += data.len() as u64;

                let mut pool = self.pool.lock();
                pool.promote_to_cached(key, Bytes::from(data));
                return Ok(true);
            }
            // Hash fail: data already removed from pool by take_all_blocks.
            return Ok(false);
        }

        // No cached data — flush and verify from disk (legacy path).
        self.flush_piece(info_hash, piece)?;
        let storage = self.get_storage(info_hash)?;
        Ok(storage.verify_piece(piece, expected)?)
    }

    fn hash_piece_v2(&self, info_hash: Id20, piece: u32, expected: &Id32) -> crate::Result<bool> {
        let key = (info_hash, piece);

        // Try hash from cache (same pattern as hash_piece but SHA-256).
        let cached_data = {
            let mut pool = self.pool.lock();
            pool.take_all_blocks(key)
        };

        if let Some(data) = cached_data {
            let hash = irontide_core::sha256(&data);
            if hash == *expected {
                let storage = self.get_storage(info_hash)?;
                storage.write_chunk(piece, 0, &data)?;
                self.stats.lock().write_bytes += data.len() as u64;

                let mut pool = self.pool.lock();
                pool.promote_to_cached(key, Bytes::from(data));
                return Ok(true);
            }
            return Ok(false);
        }

        self.flush_piece(info_hash, piece)?;
        let storage = self.get_storage(info_hash)?;
        Ok(storage.verify_piece_v2(piece, expected)?)
    }

    fn hash_block(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        length: u32,
    ) -> crate::Result<Id32> {
        self.flush_piece(info_hash, piece)?;
        let storage = self.get_storage(info_hash)?;
        Ok(storage.hash_block(piece, begin, length)?)
    }

    fn clear_piece(&self, info_hash: Id20, piece: u32) {
        self.pool.lock().clear_piece((info_hash, piece));
    }

    fn flush_piece(&self, info_hash: Id20, piece: u32) -> crate::Result<()> {
        let blocks = {
            let mut pool = self.pool.lock();
            match pool.flush_piece((info_hash, piece)) {
                Some(b) => b,
                None => return Ok(()),
            }
        };
        let storage = self.get_storage(info_hash)?;
        for (begin, data) in blocks {
            storage.write_chunk(piece, begin, &data)?;
        }
        Ok(())
    }

    fn flush_all(&self) -> crate::Result<()> {
        let all_blocks = {
            let mut pool = self.pool.lock();
            pool.flush_all()
        };
        for ((info_hash, piece), blocks) in all_blocks {
            let storage = self.get_storage(info_hash)?;
            for (begin, data) in blocks {
                storage.write_chunk(piece, begin, &data)?;
            }
        }
        Ok(())
    }

    fn cached_pieces(&self, info_hash: Id20) -> Vec<u32> {
        let pool = self.pool.lock();
        pool.hot_pieces(info_hash)
    }

    fn stats(&self) -> DiskIoStats {
        let mut s = self.stats.lock().clone();
        let pool = self.pool.lock();
        let ps = pool.stats();
        s.write_buffer_bytes = ps.write_buffer_bytes;
        s.read_cache_bytes = ps.read_cache_bytes;
        s.pool_entries = ps.total_entries;
        s.prefetch_count = ps.prefetch_count;
        s.eviction_count = ps.eviction_count;
        s.skeleton_count = ps.skeleton_count;
        s
    }

    fn write_block_direct(
        &self,
        info_hash: Id20,
        piece: u32,
        begin: u32,
        s0: &[u8],
        s1: &[u8],
    ) -> crate::Result<()> {
        let storage = self.get_storage(info_hash)?;
        storage.write_chunk_vectored(piece, begin, s0, s1)?;
        let total = (s0.len() + s1.len()) as u64;
        self.stats.lock().write_bytes += total;
        Ok(())
    }
}

// ---------------------------------------------------------------------------
// Factory
// ---------------------------------------------------------------------------

/// Create the disk I/O backend.
///
/// M257a: the universal pwritev path ([`PosixDiskIo`]) is the only production
/// backend — the optional mmap / `io_uring` / IOCP backends were deleted
/// (decision record: `docs/audits/2026-06-11-reference-architecture-gap-review.md`,
/// decision addendum). The `TorrentStorage` trait remains the seam for the
/// in-memory test backend.
#[must_use]
pub fn create_backend_from_config(config: &DiskConfig) -> Arc<dyn DiskIoBackend> {
    Arc::new(PosixDiskIo::new(config))
}

#[cfg(test)]
mod tests {
    use super::*;
    use irontide_core::Lengths;
    use irontide_storage::MemoryStorage;

    fn make_hash() -> Id20 {
        Id20([0xAB; 20])
    }

    fn make_hash_n(n: u8) -> Id20 {
        let mut b = [0u8; 20];
        b[0] = n;
        Id20(b)
    }

    fn test_config() -> DiskConfig {
        DiskConfig {
            io_threads: 2,
            cache_size: 1024 * 1024, // 1 MiB
            ..DiskConfig::default()
        }
    }

    /// Create a `MemoryStorage` with a single piece of the given size.
    fn make_storage(piece_size: u64) -> Arc<dyn TorrentStorage> {
        let chunk = piece_size.min(16384) as u32;
        let lengths = Lengths::new(piece_size, piece_size, chunk);
        Arc::new(MemoryStorage::new(lengths))
    }

    /// Create a `MemoryStorage` with the given total/piece/chunk sizes.
    fn make_storage_full(total: u64, piece_len: u64, chunk_size: u32) -> Arc<dyn TorrentStorage> {
        let lengths = Lengths::new(total, piece_len, chunk_size);
        Arc::new(MemoryStorage::new(lengths))
    }

    // -----------------------------------------------------------------------
    // DisabledDiskIo tests
    // -----------------------------------------------------------------------

    #[test]
    fn disabled_backend_name() {
        let backend = DisabledDiskIo;
        assert_eq!(backend.name(), "disabled");
    }

    #[test]
    fn disabled_backend_write_succeeds() {
        let backend = DisabledDiskIo;
        let result =
            backend.write_chunk(make_hash(), 0, 0, Bytes::from_static(&[1, 2, 3, 4]), false);
        assert!(result.is_ok());
    }

    #[test]
    fn disabled_backend_read_returns_zeroed() {
        let backend = DisabledDiskIo;
        let length = 16384u32;
        let data = backend
            .read_chunk(make_hash(), 0, 0, length, false)
            .unwrap();
        assert_eq!(data.len(), length as usize);
        assert!(data.iter().all(|&b| b == 0));
    }

    #[test]
    fn disabled_backend_hash_always_passes() {
        let backend = DisabledDiskIo;
        let expected = Id20([0xFF; 20]);
        let result = backend.hash_piece(make_hash(), 42, &expected).unwrap();
        assert!(result);
    }

    #[test]
    fn disabled_backend_hash_v2_always_passes() {
        let backend = DisabledDiskIo;
        let expected = Id32([0xFF; 32]);
        let result = backend.hash_piece_v2(make_hash(), 42, &expected).unwrap();
        assert!(result);
    }

    #[test]
    fn disabled_backend_stats_default() {
        let backend = DisabledDiskIo;
        let stats = backend.stats();
        assert_eq!(stats.read_bytes, 0);
        assert_eq!(stats.write_bytes, 0);
        assert_eq!(stats.cache_hits, 0);
        assert_eq!(stats.cache_misses, 0);
        assert_eq!(stats.write_buffer_bytes, 0);
    }

    #[test]
    fn disabled_backend_cached_pieces_empty() {
        let backend = DisabledDiskIo;
        let pieces = backend.cached_pieces(make_hash());
        assert!(pieces.is_empty());
    }

    // -----------------------------------------------------------------------
    // PosixDiskIo tests
    // -----------------------------------------------------------------------

    #[test]
    fn posix_backend_name() {
        let backend = PosixDiskIo::new(&test_config());
        assert_eq!(backend.name(), "posix");
    }

    #[test]
    fn posix_register_unregister() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(1);
        let storage = make_storage(100);
        backend.register(ih, storage);

        // Should be able to read (empty data)
        assert!(backend.read_chunk(ih, 0, 0, 10, false).is_ok());

        backend.unregister(ih);

        // Should fail after unregister
        assert!(backend.read_chunk(ih, 0, 0, 10, false).is_err());
    }

    #[test]
    fn posix_write_and_read_flush() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(2);
        let storage = make_storage(100);
        backend.register(ih, storage);

        let data = vec![42u8; 50];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data.clone()), true)
            .unwrap();
        let read = backend.read_chunk(ih, 0, 0, 50, false).unwrap();
        assert_eq!(&read[..], &data[..]);
    }

    #[test]
    fn posix_write_buffered_then_read() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(3);
        let storage = make_storage(100);
        backend.register(ih, storage);

        let data = vec![99u8; 50];
        // Write without flush — goes to buffer pool
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data.clone()), false)
            .unwrap();

        // Read should find it in buffer pool
        let read = backend.read_chunk(ih, 0, 0, 50, false).unwrap();
        assert_eq!(&read[..], &data[..]);

        // Should be a cache hit (from buffer pool)
        let stats = backend.stats();
        assert!(stats.cache_hits >= 1);
    }

    #[test]
    fn posix_read_cache_hit() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(4);
        let storage = make_storage(100);
        backend.register(ih, storage);

        let data = vec![7u8; 50];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data.clone()), true)
            .unwrap();

        // First read: cache miss, reads from storage and prefetches
        let r1 = backend.read_chunk(ih, 0, 0, 50, false).unwrap();
        assert_eq!(&r1[..], &data[..]);
        let s1 = backend.stats();
        assert_eq!(s1.cache_misses, 1);

        // Second read: should be cache hit (from prefetched Cached entry)
        let r2 = backend.read_chunk(ih, 0, 0, 50, false).unwrap();
        assert_eq!(&r2[..], &data[..]);
        let s2 = backend.stats();
        assert_eq!(s2.cache_hits, 1);
    }

    #[test]
    fn posix_hash_piece_correct() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(5);
        let storage = make_storage(50);
        backend.register(ih, storage);

        let data = vec![9u8; 50];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data.clone()), true)
            .unwrap();

        let expected = irontide_core::sha1(&data);
        assert!(backend.hash_piece(ih, 0, &expected).unwrap());
    }

    #[test]
    fn posix_hash_piece_wrong() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(6);
        let storage = make_storage(50);
        backend.register(ih, storage);

        let data = vec![9u8; 50];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data), true)
            .unwrap();

        let wrong = Id20([0xFF; 20]);
        assert!(!backend.hash_piece(ih, 0, &wrong).unwrap());
    }

    #[test]
    fn posix_hash_piece_v2() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(7);
        let storage = make_storage(16384);
        backend.register(ih, storage);

        let data = vec![0xABu8; 16384];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data.clone()), true)
            .unwrap();

        let expected = irontide_core::sha256(&data);
        assert!(backend.hash_piece_v2(ih, 0, &expected).unwrap());
    }

    #[test]
    fn posix_clear_piece_drops_buffer() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(8);
        let storage = make_storage(100);
        backend.register(ih, storage);

        let data = vec![55u8; 50];
        // Write buffered (not flushed)
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data), false)
            .unwrap();
        assert!(backend.stats().write_buffer_bytes > 0);

        backend.clear_piece(ih, 0);
        assert_eq!(backend.stats().write_buffer_bytes, 0);
    }

    #[test]
    fn posix_cached_pieces() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(9);
        // Two pieces: piece 0 and piece 1
        let storage = make_storage_full(100, 50, 25);
        backend.register(ih, storage);

        let data = vec![1u8; 25];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data.clone()), true)
            .unwrap();
        backend
            .write_chunk(ih, 1, 0, Bytes::from(data), true)
            .unwrap();

        // Read piece 0 twice to promote to T2 (hot_pieces requires ≥2 accesses)
        backend.read_chunk(ih, 0, 0, 25, false).unwrap();
        backend.read_chunk(ih, 0, 0, 25, false).unwrap();

        let cached = backend.cached_pieces(ih);
        assert!(cached.contains(&0));
        // piece 1 was not read, should not be cached
        assert!(!cached.contains(&1));
    }

    // -----------------------------------------------------------------------
    // read_piece tests
    // -----------------------------------------------------------------------

    #[test]
    fn read_piece_returns_full_piece() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(20);
        let chunk_size = 16384u32;
        let piece_size = u64::from(chunk_size) * 2; // 32768 — two chunks
        let storage = make_storage_full(piece_size, piece_size, chunk_size);
        backend.register(ih, storage);

        // Write two chunks via the backend (buffered, not flushed)
        let chunk0 = vec![0xAAu8; chunk_size as usize];
        let chunk1 = vec![0xBBu8; chunk_size as usize];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(chunk0.clone()), false)
            .expect("write chunk 0");
        backend
            .write_chunk(ih, 0, chunk_size, Bytes::from(chunk1.clone()), false)
            .expect("write chunk 1");

        // read_piece should flush the buffer pool and return the full piece
        let piece_data = backend.read_piece(ih, 0).expect("read_piece");
        assert_eq!(piece_data.len(), piece_size as usize);
        assert_eq!(&piece_data[..chunk_size as usize], &chunk0[..]);
        assert_eq!(&piece_data[chunk_size as usize..], &chunk1[..]);

        // Stats should reflect the read
        let stats = backend.stats();
        assert!(stats.read_bytes >= piece_size);
    }

    // -----------------------------------------------------------------------
    // Hash-from-cache tests
    // -----------------------------------------------------------------------

    #[test]
    fn posix_hash_from_cache_pass() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(30);
        let storage = make_storage(100);
        backend.register(ih, storage);

        // Write two blocks into the buffer pool (not flushed)
        let d1 = vec![0xAA_u8; 50];
        let d2 = vec![0xBB_u8; 50];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(d1.clone()), false)
            .unwrap();
        backend
            .write_chunk(ih, 0, 50, Bytes::from(d2.clone()), false)
            .unwrap();

        // Compute expected hash of the full piece
        let mut full = d1;
        full.extend_from_slice(&d2);
        let expected = irontide_core::sha1(&full);

        // hash_piece should hash from cache, write to disk, and promote to Cached
        assert!(backend.hash_piece(ih, 0, &expected).unwrap());

        // The piece should now be readable from disk via read_piece
        let piece_data = backend.read_piece(ih, 0).unwrap();
        assert_eq!(piece_data.len(), 100);

        // Read again to promote from T1 to T2 (hot_pieces requires ≥2 accesses)
        backend.read_chunk(ih, 0, 0, 50, false).unwrap();

        // The piece should now be in the hot pieces list (T2)
        let cached = backend.cached_pieces(ih);
        assert!(cached.contains(&0));
    }

    #[test]
    fn posix_hash_from_cache_fail() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(31);
        let storage = make_storage(50);
        backend.register(ih, storage);

        // Write one block into buffer pool
        backend
            .write_chunk(ih, 0, 0, Bytes::from(vec![0xCC_u8; 50]), false)
            .unwrap();

        // Wrong hash — should fail, data discarded from pool
        let wrong = Id20([0xFF; 20]);
        assert!(!backend.hash_piece(ih, 0, &wrong).unwrap());

        // Buffer pool should no longer hold the data
        assert_eq!(backend.stats().write_buffer_bytes, 0);
    }

    // -----------------------------------------------------------------------
    // Factory tests
    // -----------------------------------------------------------------------

    #[test]
    fn factory_creates_posix() {
        let config = DiskConfig::default();
        let backend = create_backend_from_config(&config);
        assert_eq!(backend.name(), "posix");
    }

    // -----------------------------------------------------------------------
    // Integration tests — full buffer pool lifecycle through PosixDiskIo
    // -----------------------------------------------------------------------

    #[test]
    fn piece_completion_flows_to_hash_pool() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(40);
        let storage = make_storage(100);
        backend.register(ih, storage);

        // Write two blocks (not flushed — buffered in pool)
        let d1 = vec![0xAA; 50];
        let d2 = vec![0xBB; 50];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(d1.clone()), false)
            .expect("write block 0");
        backend
            .write_chunk(ih, 0, 50, Bytes::from(d2.clone()), false)
            .expect("write block 1");

        let mut full = d1;
        full.extend_from_slice(&d2);
        let expected_hash = irontide_core::sha1(&full);

        // Hash from cache should pass
        assert!(
            backend
                .hash_piece(ih, 0, &expected_hash)
                .expect("hash_piece"),
            "hash should match"
        );

        // Data should now be on disk (read_piece flushes then reads from storage)
        let piece = backend.read_piece(ih, 0).expect("read_piece after hash");
        assert_eq!(piece, full);

        // After hash_piece, the piece was promoted to Cached. A second read
        // promotes it from T1 to T2, making it appear in hot_pieces.
        let _ = backend
            .read_chunk(ih, 0, 0, 100, false)
            .expect("second read");
        let cached = backend.cached_pieces(ih);
        assert!(
            cached.contains(&0),
            "piece 0 should be in hot pieces after two accesses"
        );
    }

    #[test]
    fn back_pressure_flush_reaches_disk() {
        // Use a tiny buffer pool (64 KiB) to force eviction under pressure.
        let config = DiskConfig {
            buffer_pool_capacity: 64 * 1024,
            ..test_config()
        };
        let backend = PosixDiskIo::new(&config);
        let ih = make_hash_n(41);
        let piece_size = 16384_u64;
        let storage = make_storage_full(piece_size * 10, piece_size, 16384);
        backend.register(ih, storage);

        // Write 8 pieces, flushing each before the next. This exercises the
        // back-pressure path: the pool never exceeds its budget because we
        // flush proactively, and every piece reaches disk.
        for p in 0..8_u32 {
            let data = vec![p as u8; piece_size as usize];
            backend
                .write_chunk(ih, p, 0, Bytes::from(data), false)
                .unwrap_or_else(|e| panic!("write piece {p}: {e}"));
            backend
                .flush_piece(ih, p)
                .unwrap_or_else(|e| panic!("flush piece {p}: {e}"));
        }

        // All 8 pieces should be on disk and readable.
        for p in 0..8_u32 {
            let piece = backend
                .read_piece(ih, p)
                .unwrap_or_else(|e| panic!("read piece {p}: {e}"));
            assert_eq!(piece.len(), piece_size as usize);
            assert!(
                piece.iter().all(|&b| b == p as u8),
                "piece {p} data mismatch"
            );
        }

        // Also verify: writing beyond budget without flushing causes eviction.
        // Write 6 more pieces without flushing (96 KiB > 64 KiB).
        for p in 0..6_u32 {
            let data = vec![(p + 10) as u8; piece_size as usize];
            backend
                .write_chunk(ih, p, 0, Bytes::from(data), false)
                .unwrap_or_else(|e| panic!("overwrite piece {p}: {e}"));
        }
        let stats = backend.stats();
        assert!(
            stats.skeleton_count > 0 || stats.eviction_count > 0,
            "should see evictions when exceeding budget: skeleton={}, eviction={}",
            stats.skeleton_count,
            stats.eviction_count
        );
    }

    #[test]
    fn prefetch_then_suggest_then_serve() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(42);
        let storage = make_storage(100);
        backend.register(ih, storage);

        // Write data and flush to disk first (simulating a completed piece on disk).
        let data = vec![0xCC; 100];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data.clone()), true)
            .expect("write flushed");

        // Read piece (triggers prefetch into cache — cache miss path).
        let r1 = backend
            .read_chunk(ih, 0, 0, 100, false)
            .expect("first read");
        assert_eq!(&r1[..], &data[..]);

        // Second read promotes to T2 in ARC.
        let r2 = backend
            .read_chunk(ih, 0, 0, 100, false)
            .expect("second read");
        assert_eq!(&r2[..], &data[..]);

        // Should appear in hot pieces now (T2).
        let cached = backend.cached_pieces(ih);
        assert!(cached.contains(&0), "piece 0 should be hot after two reads");

        // Third read should be a pure cache hit.
        let s1 = backend.stats();
        let _ = backend
            .read_chunk(ih, 0, 0, 100, false)
            .expect("third read");
        let s2 = backend.stats();
        assert!(
            s2.cache_hits > s1.cache_hits,
            "third read should be a cache hit: before={}, after={}",
            s1.cache_hits,
            s2.cache_hits
        );
    }

    #[test]
    fn full_download_verify_cycle() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(43);
        let chunk = 16384_u32;
        let piece_len = u64::from(chunk) * 4; // 4 blocks per piece
        let total = piece_len * 3; // 3 pieces
        let storage = make_storage_full(total, piece_len, chunk);
        backend.register(ih, storage);

        // Download 3 pieces: write all blocks then hash from cache.
        for p in 0..3_u32 {
            let mut full_piece = Vec::new();
            for b in 0..4_u32 {
                let block = vec![(p * 4 + b) as u8; chunk as usize];
                full_piece.extend_from_slice(&block);
                backend
                    .write_chunk(ih, p, b * chunk, Bytes::from(block), false)
                    .unwrap_or_else(|e| panic!("write piece {p} block {b}: {e}"));
            }
            let expected = irontide_core::sha1(&full_piece);
            assert!(
                backend
                    .hash_piece(ih, p, &expected)
                    .unwrap_or_else(|e| panic!("hash piece {p}: {e}")),
                "piece {p} hash should pass"
            );
        }

        // Verify all 3 pieces on disk.
        for p in 0..3_u32 {
            let data = backend
                .read_piece(ih, p)
                .unwrap_or_else(|e| panic!("read piece {p}: {e}"));
            assert_eq!(data.len(), piece_len as usize);
            for b in 0..4_u32 {
                let expected_byte = (p * 4 + b) as u8;
                let start = (b * chunk) as usize;
                let end = start + chunk as usize;
                assert!(
                    data[start..end].iter().all(|&x| x == expected_byte),
                    "piece {p} block {b}: expected 0x{expected_byte:02X}"
                );
            }
        }
    }

    #[test]
    fn skeleton_eviction_then_completion() {
        // Tiny pool: 32 KiB — forces Writing-to-Skeleton demotion.
        let config = DiskConfig {
            buffer_pool_capacity: 32 * 1024,
            ..test_config()
        };
        let backend = PosixDiskIo::new(&config);
        let ih = make_hash_n(44);
        // 4 pieces of 16 KiB each = 64 KiB > 32 KiB budget.
        let storage = make_storage_full(16384 * 4, 16384, 16384);
        backend.register(ih, storage);

        for p in 0..4_u32 {
            let data = vec![p as u8; 16384];
            backend
                .write_chunk(ih, p, 0, Bytes::from(data), false)
                .unwrap_or_else(|e| panic!("write piece {p}: {e}"));
        }

        // Flush and verify all pieces can complete via disk path.
        for p in 0..4_u32 {
            backend
                .flush_piece(ih, p)
                .unwrap_or_else(|e| panic!("flush piece {p}: {e}"));
            let piece = backend
                .read_piece(ih, p)
                .unwrap_or_else(|e| panic!("read piece {p}: {e}"));
            assert_eq!(piece.len(), 16384);
        }

        // Verify eviction/skeleton stats reflect memory pressure.
        let stats = backend.stats();
        assert!(
            stats.skeleton_count > 0 || stats.eviction_count > 0,
            "should have evictions under pressure: skeleton={}, eviction={}",
            stats.skeleton_count,
            stats.eviction_count
        );
    }

    #[test]
    fn hash_from_cache_fail_no_disk_write() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(46);
        let storage = make_storage(50);
        backend.register(ih, storage);

        let data = vec![0xDD; 50];
        backend
            .write_chunk(ih, 0, 0, Bytes::from(data), false)
            .expect("buffered write");

        // Hash with wrong expected value — should fail.
        let wrong_hash = Id20([0xFF; 20]);
        assert!(
            !backend
                .hash_piece(ih, 0, &wrong_hash)
                .expect("hash_piece should not error"),
            "hash should fail with wrong expected value"
        );

        // The piece should NOT be on disk. MemoryStorage is zero-initialized,
        // so reading back should yield all zeros (no write occurred).
        let piece = backend.read_piece(ih, 0).expect("read_piece");
        assert!(
            piece.iter().all(|&b| b == 0),
            "failed piece should not be written to disk"
        );
    }

    #[test]
    fn seeding_throughput_with_cache() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(47);
        let chunk = 16384_u32;
        let piece_len = u64::from(chunk);
        let num_pieces = 64_u32;
        let storage = make_storage_full(piece_len * u64::from(num_pieces), piece_len, chunk);
        backend.register(ih, storage);

        // Write all pieces to disk (flush=true simulates completed seeded data).
        for p in 0..num_pieces {
            let data = vec![p as u8; chunk as usize];
            backend
                .write_chunk(ih, p, 0, Bytes::from(data), true)
                .unwrap_or_else(|e| panic!("write piece {p}: {e}"));
        }

        // First read pass: cache miss + prefetch for each piece.
        for p in 0..num_pieces {
            backend
                .read_chunk(ih, p, 0, chunk, false)
                .unwrap_or_else(|e| panic!("first read piece {p}: {e}"));
        }

        // Second read pass: should be mostly cache hits.
        let stats_before = backend.stats();
        for p in 0..num_pieces {
            backend
                .read_chunk(ih, p, 0, chunk, false)
                .unwrap_or_else(|e| panic!("second read piece {p}: {e}"));
        }
        let stats_after = backend.stats();

        let hits = stats_after
            .cache_hits
            .saturating_sub(stats_before.cache_hits);
        let total = u64::from(num_pieces);
        let hit_rate = hits as f64 / total as f64;

        // Should have >80% cache hit rate with 64 x 16 KiB pieces in 1 MiB cache.
        assert!(
            hit_rate > 0.8,
            "cache hit rate {hit_rate:.1} should be >80% (hits={hits}, total={total})"
        );
    }

    // -----------------------------------------------------------------------
    // write_block_direct tests
    // -----------------------------------------------------------------------

    #[test]
    fn write_block_direct_contiguous() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(60);
        let storage = make_storage(100);
        backend.register(ih, Arc::clone(&storage));

        // Common case: all data in s0, empty s1.
        let data = vec![0xCDu8; 50];
        backend.write_block_direct(ih, 0, 0, &data, &[]).unwrap();

        let read = storage.read_chunk(0, 0, 50).unwrap();
        assert_eq!(read, data);

        // Stats should reflect the write.
        let stats = backend.stats();
        assert_eq!(stats.write_bytes, 50);
    }

    #[test]
    fn write_block_direct_split() {
        let backend = PosixDiskIo::new(&test_config());
        let ih = make_hash_n(61);
        let storage = make_storage(100);
        backend.register(ih, Arc::clone(&storage));

        // Ring-buffer wrap: 30 bytes in s0, 20 bytes in s1.
        let s0: Vec<u8> = (0..30).collect();
        let s1: Vec<u8> = (30..50).collect();
        backend.write_block_direct(ih, 0, 10, &s0, &s1).unwrap();

        let read = storage.read_chunk(0, 10, 50).unwrap();
        let expected: Vec<u8> = (0..50).collect();
        assert_eq!(read, expected);

        let stats = backend.stats();
        assert_eq!(stats.write_bytes, 50);
    }
}