libdictenstein 0.1.0

//! Buffer Manager for Persistent Adaptive Radix Trie
//!
//! This module implements a page cache with LRU eviction using the Clock algorithm.
//! It provides:
//!
//! - **Page Cache**: Fixed-size pool of in-memory pages
//! - **Clock Eviction**: O(1) amortized eviction with reference bit tracking
//! - **Read/Write Leases**: RAII guards that prevent eviction during active use
//! - **Dirty Tracking**: Pages modified in memory are tracked for write-back
//!
//! # Architecture
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────┐
//! │                    BufferManager<S>                           │
//! ├─────────────────────────────────────────────────────────────┤
//! │  Page Table: HashMap<BlockId, FrameId>                       │
//! │    Maps disk blocks to buffer pool frames                    │
//! ├─────────────────────────────────────────────────────────────┤
//! │  Frame Metadata: Vec<FrameMetadata>                          │
//! │    [frame 0] [frame 1] [frame 2] ... [frame N-1]            │
//! │    - block_id: Option<u32>                                   │
//! │    - lease_state: AtomicU32                                  │
//! │    - dirty: AtomicBool                                       │
//! │    - reference_bit: AtomicBool                               │
//! ├─────────────────────────────────────────────────────────────┤
//! │  Buffer Pool: Vec<AlignedBlock>                              │
//! │    4096-byte aligned page data (O_DIRECT compatible)        │
//! ├─────────────────────────────────────────────────────────────┤
//! │  Clock Hand: AtomicUsize                                     │
//! │    Points to next eviction candidate                        │
//! └─────────────────────────────────────────────────────────────┘
//! ```
//!
//! # Clock Algorithm
//!
//! The Clock algorithm is a practical approximation of LRU:
//!
//! 1. Each frame has a "reference bit" set on access
//! 2. Clock hand sweeps through frames looking for eviction candidates
//! 3. If reference bit is set, clear it and move on (second chance)
//! 4. If reference bit is clear and unpinned, evict that frame
//!
//! This gives O(1) amortized eviction time with good cache behavior.

use std::collections::HashMap;
use std::ptr::NonNull;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicUsize, Ordering};

use parking_lot::{Mutex, RwLock};

use super::block_storage::{AlignedBlock, BlockStorage};
use super::disk_manager::{MmapDiskManager, BLOCK_SIZE};
use super::error::{PersistentARTrieError, Result};

/// Frame ID type (index into buffer pool)
pub type FrameId = usize;

/// Metadata for a single buffer frame
#[derive(Debug)]
pub struct FrameMetadata {
    /// Block ID stored in this frame (u32::MAX = None/free)
    block_id: AtomicU32,
    /// Active lease state (frame cannot be evicted while non-zero).
    ///
    /// Values `0..WRITE_LEASE` are shared read leases. `WRITE_LEASE` is the
    /// exclusive mutable lease bit. Page write guards use an exclusive lease
    /// because `data_mut()` creates `&mut` from interior buffer storage.
    lease_state: AtomicU32,
    /// Whether the page has been modified since last write-back
    dirty: AtomicBool,
    /// Reference bit for Clock algorithm (set on access, cleared by clock hand)
    reference_bit: AtomicBool,
}

impl FrameMetadata {
    /// Sentinel value indicating no block is assigned (frame is free)
    const NONE_BLOCK: u32 = u32::MAX;
    const WRITE_LEASE: u32 = 1 << 31;
    const READERS_MASK: u32 = Self::WRITE_LEASE - 1;

    /// Create a new free frame
    fn new() -> Self {
        Self {
            block_id: AtomicU32::new(Self::NONE_BLOCK),
            lease_state: AtomicU32::new(0),
            dirty: AtomicBool::new(false),
            reference_bit: AtomicBool::new(false),
        }
    }

    /// Check if this frame is free (no block assigned)
    fn is_free(&self) -> bool {
        self.block_id.load(Ordering::Acquire) == Self::NONE_BLOCK
    }

    /// Check if this frame is pinned
    fn is_pinned(&self) -> bool {
        self.lease_state.load(Ordering::Acquire) != 0
    }

    /// Check if this frame currently has an exclusive mutable lease.
    fn has_write_lease(&self) -> bool {
        self.lease_state.load(Ordering::Acquire) & Self::WRITE_LEASE != 0
    }

    /// Acquire a shared read lease.
    fn pin_read(&self) -> Result<()> {
        loop {
            let observed = self.lease_state.load(Ordering::Acquire);
            if observed & Self::WRITE_LEASE != 0 {
                return Err(PersistentARTrieError::internal(
                    "cannot read-pin frame while an exclusive write lease is active",
                ));
            }
            if observed == Self::READERS_MASK {
                return Err(PersistentARTrieError::internal(
                    "buffer frame read lease count overflow",
                ));
            }
            if self
                .lease_state
                .compare_exchange_weak(observed, observed + 1, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
            {
                self.reference_bit.store(true, Ordering::Release);
                return Ok(());
            }
        }
    }

    /// Release a shared read lease.
    fn unpin_read(&self) {
        loop {
            let observed = self.lease_state.load(Ordering::Acquire);
            debug_assert!(
                observed > 0 && observed & Self::WRITE_LEASE == 0,
                "read unpin called without an active read lease"
            );
            if observed == 0 || observed & Self::WRITE_LEASE != 0 {
                return;
            }
            if self
                .lease_state
                .compare_exchange_weak(observed, observed - 1, Ordering::AcqRel, Ordering::Acquire)
                .is_ok()
            {
                return;
            }
        }
    }

    /// Acquire the exclusive mutable lease used by [`PageWriteGuard`].
    fn pin_write(&self) -> Result<()> {
        self.lease_state
            .compare_exchange(0, Self::WRITE_LEASE, Ordering::AcqRel, Ordering::Acquire)
            .map(|_| {
                self.reference_bit.store(true, Ordering::Release);
            })
            .map_err(|_| {
                PersistentARTrieError::internal(
                    "cannot mutably pin frame while another page lease is active",
                )
            })
    }

    /// Release the exclusive mutable lease.
    fn unpin_write(&self) {
        let result = self.lease_state.compare_exchange(
            Self::WRITE_LEASE,
            0,
            Ordering::AcqRel,
            Ordering::Acquire,
        );
        debug_assert!(
            result.is_ok(),
            "write unpin called without an active write lease"
        );
    }

    /// Mark the frame as dirty
    fn mark_dirty(&self) {
        self.dirty.store(true, Ordering::Release);
    }

    /// Check if dirty
    fn is_dirty(&self) -> bool {
        self.dirty.load(Ordering::Acquire)
    }

    /// Clear dirty flag (after write-back)
    fn clear_dirty(&self) {
        self.dirty.store(false, Ordering::Release);
    }

    /// Get the block ID
    fn get_block_id(&self) -> Option<u32> {
        match self.block_id.load(Ordering::Acquire) {
            Self::NONE_BLOCK => None,
            id => Some(id),
        }
    }

    /// Set the block ID
    fn set_block_id(&self, block_id: Option<u32>) {
        let val = block_id.unwrap_or(Self::NONE_BLOCK);
        self.block_id.store(val, Ordering::Release);
    }
}

/// RAII guard for a pinned page (read access)
///
/// The page is automatically unpinned when the guard is dropped.
pub struct PageReadGuard<'a, S: BlockStorage = MmapDiskManager> {
    buffer_manager: &'a BufferManager<S>,
    frame_id: FrameId,
}

impl<'a, S: BlockStorage> PageReadGuard<'a, S> {
    /// Get a read-only view of the page data
    pub fn data(&self) -> &[u8; BLOCK_SIZE] {
        &self.buffer_manager.buffer_pool[self.frame_id].data
    }

    /// Get the block ID of this page
    pub fn block_id(&self) -> u32 {
        self.buffer_manager.frames[self.frame_id]
            .get_block_id()
            .expect("pinned frame must have block_id")
    }
}

impl<'a, S: BlockStorage> Drop for PageReadGuard<'a, S> {
    fn drop(&mut self) {
        self.buffer_manager.frames[self.frame_id].unpin_read();
    }
}

/// RAII guard for a pinned page (write access)
///
/// The page is automatically marked dirty and unpinned when the guard is dropped.
pub struct PageWriteGuard<'a, S: BlockStorage = MmapDiskManager> {
    buffer_manager: &'a BufferManager<S>,
    frame_id: FrameId,
}

impl<'a, S: BlockStorage> PageWriteGuard<'a, S> {
    /// Get a mutable view of the page data
    ///
    /// # Safety
    /// Caller must ensure exclusive access to this frame. The buffer manager
    /// enforces this through the pinning mechanism, but the actual mutation
    /// requires unsafe due to interior mutability.
    pub fn data_mut(&mut self) -> &mut [u8; BLOCK_SIZE] {
        // Safety: We have exclusive access via the write guard and pin
        unsafe {
            let ptr = self.buffer_manager.buffer_pool.as_ptr() as *mut AlignedBlock;
            &mut (*ptr.add(self.frame_id)).data
        }
    }

    /// Get read-only view of the page data
    pub fn data(&self) -> &[u8; BLOCK_SIZE] {
        &self.buffer_manager.buffer_pool[self.frame_id].data
    }

    /// Get the block ID of this page
    pub fn block_id(&self) -> u32 {
        self.buffer_manager.frames[self.frame_id]
            .get_block_id()
            .expect("pinned frame must have block_id")
    }
}

impl<'a, S: BlockStorage> Drop for PageWriteGuard<'a, S> {
    fn drop(&mut self) {
        self.buffer_manager.frames[self.frame_id].mark_dirty();
        self.buffer_manager.frames[self.frame_id].unpin_write();
    }
}

/// Buffer manager with Clock eviction algorithm
///
/// Generic over the storage backend `S`. Defaults to `MmapDiskManager` for
/// backward compatibility — existing code using `BufferManager` without type
/// parameters continues to compile unchanged.
pub struct BufferManager<S: BlockStorage = MmapDiskManager> {
    /// The underlying storage backend
    storage: S,
    /// Page table: maps block_id -> frame_id
    page_table: RwLock<HashMap<u32, FrameId>>,
    /// Serializes frame residency changes and page-table-based lease acquisition.
    lifecycle_lock: Mutex<()>,
    /// Frame metadata
    frames: Vec<FrameMetadata>,
    /// Buffer pool (actual page data, 4096-byte aligned for O_DIRECT compatibility)
    buffer_pool: Vec<AlignedBlock>,
    /// Clock hand for eviction
    clock_hand: AtomicUsize,
    /// Maximum number of frames in the pool (allocated capacity)
    pool_size: usize,
    /// Currently active pool size (can be <= pool_size for adaptive sizing)
    active_pool_size: AtomicUsize,
    /// Whether this buffer manager's pool was successfully registered for zero-copy I/O.
    ///
    /// Set during construction if `register_buffer_pool()` succeeds. Used to decide
    /// between fixed-buffer (zero-copy) and standard I/O paths in `flush_all()`.
    fixed_buffers_registered: bool,
}

impl<S: BlockStorage> BufferManager<S> {
    /// Create a new buffer manager
    ///
    /// # Arguments
    /// * `storage` - The storage backend for I/O operations
    /// * `pool_size` - Number of frames in the buffer pool
    pub fn new(storage: S, pool_size: usize) -> Self {
        let frames: Vec<FrameMetadata> = (0..pool_size).map(|_| FrameMetadata::new()).collect();
        let buffer_pool: Vec<AlignedBlock> = (0..pool_size).map(|_| AlignedBlock::new()).collect();

        // Register buffer pool with storage backend for zero-copy I/O.
        // This is a no-op for mmap but enables ReadFixed/WriteFixed for io_uring.
        let buffers: Vec<(*mut u8, usize)> = buffer_pool
            .iter()
            .map(|block| (block.data.as_ptr() as *mut u8, BLOCK_SIZE))
            .collect();
        // Safety: buffer_pool is owned by BufferManager and will not be moved/freed
        // until BufferManager is dropped, at which point unregister_buffer_pool is called.
        // The Vec itself is never reallocated (capacity is fixed at construction).
        let fixed_buffers_registered = unsafe { storage.register_buffer_pool(&buffers).is_ok() }
            && storage.supports_fixed_buffers();

        Self {
            storage,
            page_table: RwLock::new(HashMap::with_capacity(pool_size)),
            lifecycle_lock: Mutex::new(()),
            frames,
            buffer_pool,
            clock_hand: AtomicUsize::new(0),
            pool_size,
            active_pool_size: AtomicUsize::new(pool_size),
            fixed_buffers_registered,
        }
    }

    /// Create a buffer manager without registering buffers for zero-copy I/O.
    ///
    /// Used for benchmarking to isolate the effect of pre-registered buffers.
    /// In production, use [`new()`](Self::new) which always attempts registration.
    ///
    /// # Arguments
    /// * `storage` - The storage backend for I/O operations
    /// * `pool_size` - Number of frames in the buffer pool
    #[cfg(any(test, feature = "bench-internals"))]
    pub fn new_without_registration(storage: S, pool_size: usize) -> Self {
        let frames: Vec<FrameMetadata> = (0..pool_size).map(|_| FrameMetadata::new()).collect();
        let buffer_pool: Vec<AlignedBlock> = (0..pool_size).map(|_| AlignedBlock::new()).collect();

        Self {
            storage,
            page_table: RwLock::new(HashMap::with_capacity(pool_size)),
            lifecycle_lock: Mutex::new(()),
            frames,
            buffer_pool,
            clock_hand: AtomicUsize::new(0),
            pool_size,
            active_pool_size: AtomicUsize::new(pool_size),
            fixed_buffers_registered: false,
        }
    }

    /// Create a new buffer manager with adaptive sizing support.
    ///
    /// Pre-allocates `max_pool_size` frames but starts with only `initial_size` active.
    /// Use `grow_pool()` and `shrink_pool()` to adjust the active pool size.
    ///
    /// # Arguments
    /// * `storage` - The storage backend for I/O operations
    /// * `initial_size` - Initial number of active frames
    /// * `max_pool_size` - Maximum number of frames (pre-allocated)
    pub fn new_with_max_capacity(storage: S, initial_size: usize, max_pool_size: usize) -> Self {
        let frames: Vec<FrameMetadata> = (0..max_pool_size).map(|_| FrameMetadata::new()).collect();
        let buffer_pool: Vec<AlignedBlock> =
            (0..max_pool_size).map(|_| AlignedBlock::new()).collect();

        // Register buffer pool with storage backend for zero-copy I/O.
        // Registers ALL max_pool_size buffers (not just initial_size) since the
        // registration covers the full pre-allocated capacity.
        let buffers: Vec<(*mut u8, usize)> = buffer_pool
            .iter()
            .map(|block| (block.data.as_ptr() as *mut u8, BLOCK_SIZE))
            .collect();
        // Safety: buffer_pool is owned by BufferManager and will not be moved/freed
        // until BufferManager is dropped, at which point unregister_buffer_pool is called.
        let fixed_buffers_registered = unsafe { storage.register_buffer_pool(&buffers).is_ok() }
            && storage.supports_fixed_buffers();

        Self {
            storage,
            page_table: RwLock::new(HashMap::with_capacity(max_pool_size)),
            lifecycle_lock: Mutex::new(()),
            frames,
            buffer_pool,
            clock_hand: AtomicUsize::new(0),
            pool_size: max_pool_size,
            active_pool_size: AtomicUsize::new(initial_size.min(max_pool_size)),
            fixed_buffers_registered,
        }
    }

    /// Fetch a page for reading
    ///
    /// If the page is already in the buffer pool, returns a guard immediately.
    /// Otherwise, loads the page from disk (potentially evicting another page).
    ///
    /// # Lease contention
    ///
    /// The returned [`PageReadGuard`] holds a shared **read lease**. Many
    /// readers may hold read leases on the same page concurrently, but this
    /// call returns an error if an exclusive write lease is currently held on
    /// the page (see [`fetch_page_mut`](Self::fetch_page_mut)).
    pub fn fetch_page(&self, block_id: u32) -> Result<PageReadGuard<'_, S>> {
        let _lifecycle = self.lifecycle_lock.lock();

        // Check if already in buffer pool
        if let Some(frame_id) = self.lookup_frame(block_id) {
            self.frames[frame_id].pin_read()?;
            self.frames[frame_id]
                .reference_bit
                .store(true, Ordering::Release);
            return Ok(PageReadGuard {
                buffer_manager: self,
                frame_id,
            });
        }

        // Need to load from disk
        let frame_id = self.load_page(block_id)?;
        Ok(PageReadGuard {
            buffer_manager: self,
            frame_id,
        })
    }

    /// Fetch a page for writing
    ///
    /// Similar to `fetch_page`, but the returned guard will mark the page
    /// dirty when dropped.
    ///
    /// # Lease contention
    ///
    /// The returned [`PageWriteGuard`] holds an **exclusive** write lease: this
    /// call returns an error if any other lease (read or write) is currently
    /// held on the page. Callers that may race other accessors of the same page
    /// must be prepared to retry. Internal callers serialize through the outer
    /// `RwLock<BufferManager>` (writers take the write side), so they never
    /// observe contention here.
    pub fn fetch_page_mut(&self, block_id: u32) -> Result<PageWriteGuard<'_, S>> {
        let _lifecycle = self.lifecycle_lock.lock();

        // Check if already in buffer pool
        if let Some(frame_id) = self.lookup_frame(block_id) {
            self.frames[frame_id].pin_write()?;
            self.frames[frame_id]
                .reference_bit
                .store(true, Ordering::Release);
            return Ok(PageWriteGuard {
                buffer_manager: self,
                frame_id,
            });
        }

        // Need to load from disk
        let frame_id = self.load_page_mut(block_id)?;
        Ok(PageWriteGuard {
            buffer_manager: self,
            frame_id,
        })
    }

    /// Create a new page (allocate a new block)
    ///
    /// Returns a write guard for the newly allocated page.
    pub fn new_page(&self) -> Result<PageWriteGuard<'_, S>> {
        // Allocate a new block on disk
        let block_id = self.storage.allocate_block()?;

        let _lifecycle = self.lifecycle_lock.lock();

        // Get a frame for it. On failure (pool exhausted, etc.) free the
        // just-allocated block so it is not leaked on disk.
        let frame_id = match self.get_free_frame() {
            Ok(frame_id) => frame_id,
            Err(e) => {
                let _ = self.storage.free_block(block_id);
                return Err(e);
            }
        };

        // Initialize the frame
        self.frames[frame_id].set_block_id(Some(block_id));
        if let Err(e) = self.frames[frame_id].pin_write() {
            // The frame was free (no block, unpinned) so this is effectively
            // unreachable, but roll back the block_id assignment and free the
            // block defensively so neither the frame nor the block is leaked.
            self.frames[frame_id].set_block_id(None);
            let _ = self.storage.free_block(block_id);
            return Err(e);
        }
        self.frames[frame_id].mark_dirty();

        // Clear the buffer
        // Safety: We have exclusive access via the new allocation
        unsafe {
            let ptr = self.buffer_pool.as_ptr() as *mut AlignedBlock;
            (*ptr.add(frame_id)).data.fill(0);
        }

        // Update page table
        self.page_table.write().insert(block_id, frame_id);

        Ok(PageWriteGuard {
            buffer_manager: self,
            frame_id,
        })
    }

    /// Delete a page
    ///
    /// The page must not be pinned by anyone else.
    pub fn delete_page(&self, block_id: u32) -> Result<()> {
        let _lifecycle = self.lifecycle_lock.lock();

        // Check if in buffer pool
        if let Some(frame_id) = self.lookup_frame(block_id) {
            let frame = &self.frames[frame_id];

            // Can't delete a pinned page
            if frame.is_pinned() {
                return Err(PersistentARTrieError::InternalError {
                    message: format!("Cannot delete pinned page (block {})", block_id),
                });
            }

            // Clear the frame
            frame.set_block_id(None);
            frame.clear_dirty();
            frame.reference_bit.store(false, Ordering::Release);

            // Remove from page table
            self.page_table.write().remove(&block_id);
        }

        // Free the block on disk
        self.storage.free_block(block_id)
    }

    /// Flush a specific page to disk
    pub fn flush_page(&self, block_id: u32) -> Result<()> {
        let _lifecycle = self.lifecycle_lock.lock();

        if let Some(frame_id) = self.lookup_frame(block_id) {
            let frame = &self.frames[frame_id];

            if frame.is_dirty() {
                if frame.has_write_lease() {
                    return Err(PersistentARTrieError::internal(
                        "cannot flush page during an active mutable page lease",
                    ));
                }
                self.storage.write_block_fixed(
                    block_id,
                    &self.buffer_pool[frame_id].data,
                    frame_id as u16,
                )?;
                frame.clear_dirty();
            }
        }
        Ok(())
    }

    /// Flush all dirty pages to disk via batched I/O.
    ///
    /// Collects all dirty frames and writes them in a single batch operation,
    /// using zero-copy fixed buffers when available (io_uring backend) or
    /// standard batched writes otherwise. Dirty flags are cleared only AFTER
    /// the batch write succeeds, ensuring retry safety on failure.
    pub fn flush_all(&self) -> Result<()> {
        let _lifecycle = self.lifecycle_lock.lock();
        let active_size = self.active_pool_size.load(Ordering::Acquire);

        // Collect all dirty (block_id, frame_id) pairs within active pool
        let dirty_frames: Vec<(u32, usize)> = self.frames[..active_size]
            .iter()
            .enumerate()
            .filter_map(|(frame_id, frame)| {
                if frame.is_dirty() {
                    frame.get_block_id().map(|block_id| (block_id, frame_id))
                } else {
                    None
                }
            })
            .collect();

        if !dirty_frames.is_empty() {
            if dirty_frames
                .iter()
                .any(|&(_, frame_id)| self.frames[frame_id].has_write_lease())
            {
                return Err(PersistentARTrieError::internal(
                    "cannot flush dirty pages during an active mutable page lease",
                ));
            }

            if self.fixed_buffers_registered {
                // Zero-copy batched flush via pre-registered buffers
                let requests: Vec<(u32, &[u8; BLOCK_SIZE], u16)> = dirty_frames
                    .iter()
                    .map(|&(block_id, frame_id)| {
                        (block_id, &self.buffer_pool[frame_id].data, frame_id as u16)
                    })
                    .collect();
                self.storage.write_blocks_batch_fixed(&requests)?;
            } else {
                // Fallback: batch without fixed buffers (still batches SQEs for io_uring)
                let requests: Vec<(u32, &[u8; BLOCK_SIZE])> = dirty_frames
                    .iter()
                    .map(|&(block_id, frame_id)| (block_id, &self.buffer_pool[frame_id].data))
                    .collect();
                self.storage.write_blocks_batch(&requests)?;
            }

            // Clear dirty flags AFTER successful batch write (retry safety)
            for &(_, frame_id) in &dirty_frames {
                self.frames[frame_id].clear_dirty();
            }
        }

        self.storage.sync()
    }

    /// Look up a frame by block ID
    fn lookup_frame(&self, block_id: u32) -> Option<FrameId> {
        self.page_table.read().get(&block_id).copied()
    }

    /// Load a page from disk into a frame
    fn load_page(&self, block_id: u32) -> Result<FrameId> {
        // Get a free frame (may evict)
        let frame_id = self.get_free_frame()?;

        // Read from disk using zero-copy path if available
        // Safety: We have exclusive access to this frame via get_free_frame
        unsafe {
            let ptr = self.buffer_pool.as_ptr() as *mut AlignedBlock;
            self.storage.read_block_fixed(
                block_id,
                &mut (*ptr.add(frame_id)).data,
                frame_id as u16,
            )?;
        }

        // Set up the frame
        self.frames[frame_id].set_block_id(Some(block_id));
        self.frames[frame_id].pin_read()?;
        self.frames[frame_id].clear_dirty();
        self.frames[frame_id]
            .reference_bit
            .store(true, Ordering::Release);

        // Update page table
        self.page_table.write().insert(block_id, frame_id);

        Ok(frame_id)
    }

    /// Load a page from disk into a frame and return it with an exclusive write lease.
    fn load_page_mut(&self, block_id: u32) -> Result<FrameId> {
        let frame_id = self.get_free_frame()?;

        // Read from disk before granting mutable access to the frame contents.
        // Safety: We have exclusive access to this frame via get_free_frame.
        unsafe {
            let ptr = self.buffer_pool.as_ptr() as *mut AlignedBlock;
            self.storage.read_block_fixed(
                block_id,
                &mut (*ptr.add(frame_id)).data,
                frame_id as u16,
            )?;
        }

        self.frames[frame_id].set_block_id(Some(block_id));
        self.frames[frame_id].pin_write()?;
        self.frames[frame_id].clear_dirty();
        self.frames[frame_id]
            .reference_bit
            .store(true, Ordering::Release);

        self.page_table.write().insert(block_id, frame_id);

        Ok(frame_id)
    }

    /// Pin a page for raw-pointer-backed traversal caching.
    ///
    /// The returned pointer remains valid until [`Self::unpin_read_frame`] is
    /// called for the frame, or until the buffer manager is dropped.
    pub(crate) fn pin_page_data(
        &self,
        block_id: u32,
    ) -> Result<(FrameId, NonNull<[u8; BLOCK_SIZE]>)> {
        let _lifecycle = self.lifecycle_lock.lock();

        let frame_id = if let Some(frame_id) = self.lookup_frame(block_id) {
            self.frames[frame_id].pin_read()?;
            self.frames[frame_id]
                .reference_bit
                .store(true, Ordering::Release);
            frame_id
        } else {
            self.load_page(block_id)?
        };

        Ok((frame_id, NonNull::from(&self.buffer_pool[frame_id].data)))
    }

    /// Release a read lease acquired by [`Self::pin_page_data`].
    pub(crate) fn unpin_read_frame(&self, frame_id: FrameId) {
        self.frames[frame_id].unpin_read();
    }

    /// Get a free frame using the Clock algorithm
    fn get_free_frame(&self) -> Result<FrameId> {
        let active_size = self.active_pool_size.load(Ordering::Acquire);

        // First pass: look for a free frame within active pool
        for frame_id in 0..active_size {
            if self.frames[frame_id].is_free() && !self.frames[frame_id].is_pinned() {
                return Ok(frame_id);
            }
        }

        // No free frames, need to evict using Clock algorithm
        let mut attempts = 0;
        let max_attempts = active_size * 2; // Two full sweeps

        while attempts < max_attempts {
            let frame_id = self.clock_hand.fetch_add(1, Ordering::Relaxed) % active_size;
            let frame = &self.frames[frame_id];

            // Skip pinned frames
            if frame.is_pinned() {
                attempts += 1;
                continue;
            }

            // Check reference bit
            if frame.reference_bit.swap(false, Ordering::AcqRel) {
                // Reference bit was set, give second chance
                attempts += 1;
                continue;
            }

            // Found a victim: evict it
            if let Some(old_block_id) = frame.get_block_id() {
                // Write back if dirty (zero-copy path if available)
                if frame.is_dirty() {
                    self.storage.write_block_fixed(
                        old_block_id,
                        &self.buffer_pool[frame_id].data,
                        frame_id as u16,
                    )?;
                    frame.clear_dirty();
                }

                // Remove from page table
                self.page_table.write().remove(&old_block_id);

                // Clear the frame
                frame.set_block_id(None);
            }

            return Ok(frame_id);
        }

        // All frames are pinned
        Err(PersistentARTrieError::BufferPoolExhausted {
            pinned_pages: self.count_pinned(),
            total_pages: active_size,
        })
    }

    /// Count the number of pinned pages
    fn count_pinned(&self) -> usize {
        self.frames.iter().filter(|f| f.is_pinned()).count()
    }

    /// Get the current active pool size.
    ///
    /// This is the number of frames currently available for use.
    /// May be less than `max_pool_size()` if using adaptive sizing.
    pub fn pool_size(&self) -> usize {
        self.active_pool_size.load(Ordering::Relaxed)
    }

    /// Get the maximum pool size (allocated capacity).
    ///
    /// This is the total number of pre-allocated frames.
    /// The active pool size can grow up to this limit.
    pub fn max_pool_size(&self) -> usize {
        self.pool_size
    }

    /// Grow the buffer pool by activating more pre-allocated frames.
    ///
    /// # Arguments
    /// * `additional_frames` - Number of frames to activate
    ///
    /// # Returns
    /// Ok(new_size) on success, Err if would exceed max capacity.
    ///
    /// # Note
    /// This only works if the BufferManager was created with
    /// `new_with_max_capacity()`. Growing beyond the pre-allocated
    /// capacity is not supported.
    pub fn grow_pool(&self, additional_frames: usize) -> Result<usize> {
        let _lifecycle = self.lifecycle_lock.lock();

        loop {
            let current = self.active_pool_size.load(Ordering::Acquire);
            let new_size = current.saturating_add(additional_frames);

            // Cannot exceed pre-allocated capacity
            if new_size > self.pool_size {
                return Err(PersistentARTrieError::InternalError {
                    message: format!(
                        "Cannot grow pool beyond max capacity {} (current: {}, requested: +{})",
                        self.pool_size, current, additional_frames
                    ),
                });
            }

            // Try to update atomically
            match self.active_pool_size.compare_exchange(
                current,
                new_size,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => return Ok(new_size),
                Err(_) => continue, // Retry
            }
        }
    }

    /// Shrink the buffer pool by deactivating frames.
    ///
    /// # Arguments
    /// * `frames_to_remove` - Number of frames to deactivate
    ///
    /// # Returns
    /// Ok(new_size) on success, Err if not enough frames or frames in use.
    ///
    /// # Note
    /// Frames that are pinned or contain data must be flushed first.
    /// This method will evict unpinned frames in the shrink range.
    pub fn shrink_pool(&self, frames_to_remove: usize) -> Result<usize> {
        let _lifecycle = self.lifecycle_lock.lock();

        loop {
            let current = self.active_pool_size.load(Ordering::Acquire);

            // Minimum pool size of 1
            let new_size = current.saturating_sub(frames_to_remove).max(1);

            if new_size == current {
                return Ok(current); // Nothing to shrink
            }

            // Check that frames in the shrink range are not pinned
            for frame_id in new_size..current {
                let frame = &self.frames[frame_id];
                if frame.is_pinned() {
                    return Err(PersistentARTrieError::InternalError {
                        message: format!("Cannot shrink pool: frame {} is pinned", frame_id),
                    });
                }

                // Flush dirty frames before shrinking (zero-copy path if available)
                if frame.is_dirty() {
                    if let Some(block_id) = frame.get_block_id() {
                        self.storage.write_block_fixed(
                            block_id,
                            &self.buffer_pool[frame_id].data,
                            frame_id as u16,
                        )?;
                        frame.clear_dirty();
                    }
                }

                // Evict the frame
                if let Some(block_id) = frame.get_block_id() {
                    self.page_table.write().remove(&block_id);
                    frame.set_block_id(None);
                }
            }

            // Try to update atomically
            match self.active_pool_size.compare_exchange(
                current,
                new_size,
                Ordering::AcqRel,
                Ordering::Acquire,
            ) {
                Ok(_) => {
                    // Reset clock hand if it's beyond new size
                    let clock = self.clock_hand.load(Ordering::Relaxed);
                    if clock >= new_size {
                        self.clock_hand.store(0, Ordering::Relaxed);
                    }
                    return Ok(new_size);
                }
                Err(_) => continue, // Retry
            }
        }
    }

    /// Get statistics about the buffer pool
    pub fn stats(&self) -> BufferPoolStats {
        let active_size = self.active_pool_size.load(Ordering::Relaxed);
        let mut free = 0;
        let mut pinned = 0;
        let mut dirty = 0;

        // Only count frames within active pool
        for frame_id in 0..active_size {
            let frame = &self.frames[frame_id];
            if frame.is_free() {
                free += 1;
            } else {
                if frame.is_pinned() {
                    pinned += 1;
                }
                if frame.is_dirty() {
                    dirty += 1;
                }
            }
        }

        BufferPoolStats {
            total_frames: active_size,
            max_frames: self.pool_size,
            free_frames: free,
            pinned_frames: pinned,
            dirty_frames: dirty,
            used_frames: active_size - free,
        }
    }

    /// Get a reference to the underlying storage backend.
    pub fn storage(&self) -> &S {
        &self.storage
    }

    /// Get a reference to the underlying storage backend.
    ///
    /// **Deprecated**: Use [`storage()`](Self::storage) instead.
    /// Retained for backward compatibility during the mmap-to-io_uring migration.
    #[deprecated(since = "0.9.0", note = "Use storage() instead")]
    pub fn disk_manager(&self) -> &S {
        &self.storage
    }
}

impl<S: BlockStorage> Drop for BufferManager<S> {
    fn drop(&mut self) {
        // Unregister the buffer pool from the storage backend.
        // This must happen before the buffer_pool Vec is dropped to avoid
        // use-after-free of the registered buffer pointers.
        let _ = self.storage.unregister_buffer_pool();
    }
}

/// Statistics about the buffer pool
#[derive(Debug, Clone, Copy)]
pub struct BufferPoolStats {
    /// Total number of active frames in the pool
    pub total_frames: usize,
    /// Maximum number of frames (allocated capacity)
    pub max_frames: usize,
    /// Number of free (unallocated) frames
    pub free_frames: usize,
    /// Number of pinned frames (cannot be evicted)
    pub pinned_frames: usize,
    /// Number of dirty frames (need write-back)
    pub dirty_frames: usize,
    /// Number of frames with data (total - free)
    pub used_frames: usize,
}

#[cfg(test)]
mod tests {
    use super::super::disk_manager::{DiskManager, FileHeader};
    use super::*;
    use std::sync::atomic::AtomicBool;
    use std::sync::Arc;
    use tempfile::tempdir;

    fn create_buffer_manager(pool_size: usize) -> BufferManager {
        let dir = tempdir().expect("Failed to create temp dir");
        let path = dir.path().join("test.part");
        let disk_manager = DiskManager::create(&path).expect("Failed to create disk manager");

        // Keep the temp dir alive by leaking it (for tests only)
        std::mem::forget(dir);

        BufferManager::new(disk_manager, pool_size)
    }

    #[derive(Clone)]
    struct TrackingFixedStorage {
        state: Arc<TrackingFixedStorageState>,
    }

    struct TrackingFixedStorageState {
        blocks: RwLock<Vec<Box<AlignedBlock>>>,
        registered: AtomicBool,
        register_calls: AtomicUsize,
        unregister_calls: AtomicUsize,
        registered_buffers: AtomicUsize,
        regular_writes: AtomicUsize,
        fixed_writes: AtomicUsize,
        fixed_batch_writes: AtomicUsize,
        fixed_reads: AtomicUsize,
    }

    impl TrackingFixedStorage {
        fn new() -> Self {
            Self {
                state: Arc::new(TrackingFixedStorageState {
                    blocks: RwLock::new(vec![AlignedBlock::new_boxed()]),
                    registered: AtomicBool::new(false),
                    register_calls: AtomicUsize::new(0),
                    unregister_calls: AtomicUsize::new(0),
                    registered_buffers: AtomicUsize::new(0),
                    regular_writes: AtomicUsize::new(0),
                    fixed_writes: AtomicUsize::new(0),
                    fixed_batch_writes: AtomicUsize::new(0),
                    fixed_reads: AtomicUsize::new(0),
                }),
            }
        }
    }

    impl TrackingFixedStorageState {
        fn invalid_block(block_id: u32) -> PersistentARTrieError {
            PersistentARTrieError::InvalidBlockId {
                block_id,
                reason: "tracking storage block does not exist".to_string(),
            }
        }

        fn invalid_range(block_id: u32) -> PersistentARTrieError {
            PersistentARTrieError::InvalidBlockId {
                block_id,
                reason: "tracking storage byte range is outside the block".to_string(),
            }
        }

        fn read_block_into(&self, block_id: u32, buffer: &mut [u8; BLOCK_SIZE]) -> Result<()> {
            let blocks = self.blocks.read();
            let block = blocks
                .get(block_id as usize)
                .ok_or_else(|| Self::invalid_block(block_id))?;
            buffer.copy_from_slice(&block.data);
            Ok(())
        }

        fn write_block_from(&self, block_id: u32, buffer: &[u8; BLOCK_SIZE]) -> Result<()> {
            let mut blocks = self.blocks.write();
            let block = blocks
                .get_mut(block_id as usize)
                .ok_or_else(|| Self::invalid_block(block_id))?;
            block.data.copy_from_slice(buffer);
            Ok(())
        }

        fn read_bytes_into(&self, block_id: u32, offset: usize, buffer: &mut [u8]) -> Result<()> {
            let end = offset
                .checked_add(buffer.len())
                .ok_or_else(|| Self::invalid_range(block_id))?;
            if end > BLOCK_SIZE {
                return Err(Self::invalid_range(block_id));
            }

            let blocks = self.blocks.read();
            let block = blocks
                .get(block_id as usize)
                .ok_or_else(|| Self::invalid_block(block_id))?;
            buffer.copy_from_slice(&block.data[offset..end]);
            Ok(())
        }

        fn write_bytes_from(&self, block_id: u32, offset: usize, data: &[u8]) -> Result<()> {
            let end = offset
                .checked_add(data.len())
                .ok_or_else(|| Self::invalid_range(block_id))?;
            if end > BLOCK_SIZE {
                return Err(Self::invalid_range(block_id));
            }

            let mut blocks = self.blocks.write();
            let block = blocks
                .get_mut(block_id as usize)
                .ok_or_else(|| Self::invalid_block(block_id))?;
            block.data[offset..end].copy_from_slice(data);
            Ok(())
        }
    }

    impl BlockStorage for TrackingFixedStorage {
        fn read_block(&self, block_id: u32, buffer: &mut [u8; BLOCK_SIZE]) -> Result<()> {
            self.state.read_block_into(block_id, buffer)
        }

        fn write_block(&self, block_id: u32, buffer: &[u8; BLOCK_SIZE]) -> Result<()> {
            self.state.regular_writes.fetch_add(1, Ordering::AcqRel);
            self.state.write_block_from(block_id, buffer)
        }

        fn read_bytes(&self, block_id: u32, offset: usize, buffer: &mut [u8]) -> Result<()> {
            self.state.read_bytes_into(block_id, offset, buffer)
        }

        fn write_bytes(&self, block_id: u32, offset: usize, data: &[u8]) -> Result<()> {
            self.state.write_bytes_from(block_id, offset, data)
        }

        fn allocate_block(&self) -> Result<u32> {
            let mut blocks = self.state.blocks.write();
            let block_id = blocks.len() as u32;
            blocks.push(AlignedBlock::new_boxed());
            Ok(block_id)
        }

        fn free_block(&self, block_id: u32) -> Result<()> {
            if block_id == 0 {
                return Err(TrackingFixedStorageState::invalid_block(block_id));
            }

            let mut blocks = self.state.blocks.write();
            let block = blocks
                .get_mut(block_id as usize)
                .ok_or_else(|| TrackingFixedStorageState::invalid_block(block_id))?;
            block.data.fill(0);
            Ok(())
        }

        fn read_header(&self) -> Result<FileHeader> {
            let mut bytes = [0u8; 64];
            self.read_header_bytes(&mut bytes)?;
            Ok(FileHeader::from_bytes(&bytes))
        }

        fn write_header(&self, header: &FileHeader) -> Result<()> {
            self.write_header_bytes(&header.to_bytes())
        }

        fn read_header_bytes(&self, buffer: &mut [u8]) -> Result<()> {
            self.state.read_bytes_into(0, 0, buffer)
        }

        fn write_header_bytes(&self, bytes: &[u8]) -> Result<()> {
            self.state.write_bytes_from(0, 0, bytes)
        }

        fn root_ptr(&self) -> Result<u64> {
            Ok(self.read_header()?.root_ptr.load(Ordering::SeqCst))
        }

        fn set_root_ptr(&self, ptr: u64) -> Result<()> {
            let header = self.read_header()?;
            header.root_ptr.store(ptr, Ordering::SeqCst);
            self.write_header(&header)
        }

        fn entry_count(&self) -> Result<u64> {
            Ok(self.read_header()?.entry_count.load(Ordering::SeqCst))
        }

        fn set_entry_count(&self, count: u64) -> Result<()> {
            let header = self.read_header()?;
            header.entry_count.store(count, Ordering::SeqCst);
            self.write_header(&header)
        }

        fn file_size(&self) -> u64 {
            self.state.blocks.read().len() as u64 * BLOCK_SIZE as u64
        }

        fn block_count(&self) -> Result<u32> {
            Ok(self.state.blocks.read().len() as u32)
        }

        fn path(&self) -> &str {
            "tracking-fixed-storage"
        }

        fn sync(&self) -> Result<()> {
            Ok(())
        }

        unsafe fn register_buffer_pool(&self, buffers: &[(*mut u8, usize)]) -> Result<()> {
            assert!(
                buffers
                    .iter()
                    .all(|(ptr, len)| !ptr.is_null() && *len == BLOCK_SIZE),
                "registered buffers must be non-null full blocks"
            );
            self.state.register_calls.fetch_add(1, Ordering::AcqRel);
            self.state
                .registered_buffers
                .store(buffers.len(), Ordering::Release);
            self.state.registered.store(true, Ordering::Release);
            Ok(())
        }

        fn unregister_buffer_pool(&self) -> Result<()> {
            self.state.unregister_calls.fetch_add(1, Ordering::AcqRel);
            self.state.registered.store(false, Ordering::Release);
            Ok(())
        }

        fn read_block_fixed(
            &self,
            block_id: u32,
            buffer: &mut [u8; BLOCK_SIZE],
            _buf_index: u16,
        ) -> Result<()> {
            self.state.fixed_reads.fetch_add(1, Ordering::AcqRel);
            self.state.read_block_into(block_id, buffer)
        }

        fn write_block_fixed(
            &self,
            block_id: u32,
            buffer: &[u8; BLOCK_SIZE],
            _buf_index: u16,
        ) -> Result<()> {
            self.state.fixed_writes.fetch_add(1, Ordering::AcqRel);
            self.state.write_block_from(block_id, buffer)
        }

        fn supports_fixed_buffers(&self) -> bool {
            self.state.registered.load(Ordering::Acquire)
        }

        fn write_blocks_batch_fixed(
            &self,
            requests: &[(u32, &[u8; BLOCK_SIZE], u16)],
        ) -> Result<()> {
            self.state.fixed_batch_writes.fetch_add(1, Ordering::AcqRel);
            for &(block_id, buffer, buf_index) in requests {
                self.write_block_fixed(block_id, buffer, buf_index)?;
            }
            Ok(())
        }
    }

    #[test]
    fn fixed_buffer_registration_covers_write_guard_mutation_and_flush() {
        let storage = TrackingFixedStorage::new();
        let state = Arc::clone(&storage.state);
        let block_id;

        {
            let bm = BufferManager::new(storage, 2);
            assert_eq!(state.register_calls.load(Ordering::Acquire), 1);
            assert_eq!(state.registered_buffers.load(Ordering::Acquire), 2);
            assert!(state.registered.load(Ordering::Acquire));

            {
                let mut guard = bm.new_page().expect("new page");
                block_id = guard.block_id();
                guard.data_mut()[7] = 0xA5;
                guard.data_mut()[BLOCK_SIZE - 1] = 0x5A;
            }

            assert_eq!(bm.stats().dirty_frames, 1);
            bm.flush_all().expect("flush fixed buffer page");
            assert_eq!(bm.stats().dirty_frames, 0);
            assert_eq!(state.regular_writes.load(Ordering::Acquire), 0);
            assert_eq!(state.fixed_batch_writes.load(Ordering::Acquire), 1);
            assert_eq!(state.fixed_writes.load(Ordering::Acquire), 1);

            let blocks = state.blocks.read();
            assert_eq!(blocks[block_id as usize].data[7], 0xA5);
            assert_eq!(blocks[block_id as usize].data[BLOCK_SIZE - 1], 0x5A);
        }

        assert_eq!(state.unregister_calls.load(Ordering::Acquire), 1);
        assert!(!state.registered.load(Ordering::Acquire));
    }

    #[test]
    fn test_new_page() {
        let bm = create_buffer_manager(10);

        let mut guard = bm.new_page().expect("new_page");
        let block_id = guard.block_id();

        // Write some data
        guard.data_mut()[0] = 0xDE;
        guard.data_mut()[1] = 0xAD;
        drop(guard);

        // Read it back
        let guard = bm.fetch_page(block_id).expect("fetch_page");
        assert_eq!(guard.data()[0], 0xDE);
        assert_eq!(guard.data()[1], 0xAD);
    }

    #[test]
    fn test_fetch_page() {
        let bm = create_buffer_manager(10);

        // Create a page and write data
        let mut guard = bm.new_page().expect("new_page");
        let block_id = guard.block_id();
        guard.data_mut()[100] = 42;
        drop(guard);

        // Flush to disk
        bm.flush_page(block_id).expect("flush");

        // Fetch again and verify
        let guard = bm.fetch_page(block_id).expect("fetch_page");
        assert_eq!(guard.data()[100], 42);
    }

    #[test]
    fn mutable_page_lease_excludes_other_leases() {
        let bm = create_buffer_manager(10);

        let mut guard = bm.new_page().expect("new_page");
        let block_id = guard.block_id();
        guard.data_mut()[0] = 7;
        drop(guard);

        let mut write_guard = bm.fetch_page_mut(block_id).expect("fetch_page_mut");
        write_guard.data_mut()[0] = 9;

        assert!(
            bm.fetch_page(block_id).is_err(),
            "read lease must not overlap an active mutable page lease"
        );
        assert!(
            bm.fetch_page_mut(block_id).is_err(),
            "second mutable lease must not alias the first mutable page lease"
        );

        drop(write_guard);

        let read_guard = bm.fetch_page(block_id).expect("fetch after write lease");
        assert_eq!(read_guard.data()[0], 9);
    }

    #[test]
    fn read_page_lease_blocks_mutable_lease() {
        let bm = create_buffer_manager(10);

        let mut guard = bm.new_page().expect("new_page");
        let block_id = guard.block_id();
        guard.data_mut()[0] = 11;
        drop(guard);

        let read_guard = bm.fetch_page(block_id).expect("fetch_page");
        assert_eq!(read_guard.data()[0], 11);
        assert!(
            bm.fetch_page_mut(block_id).is_err(),
            "mutable page lease must not overlap an active read lease"
        );

        drop(read_guard);

        let mut write_guard = bm
            .fetch_page_mut(block_id)
            .expect("fetch_page_mut after read");
        write_guard.data_mut()[0] = 12;
    }

    #[test]
    fn flush_rejects_dirty_page_with_active_mutable_lease() {
        let bm = create_buffer_manager(10);

        let mut guard = bm.new_page().expect("new_page");
        let block_id = guard.block_id();
        guard.data_mut()[0] = 1;
        drop(guard);

        let mut write_guard = bm.fetch_page_mut(block_id).expect("fetch_page_mut");
        write_guard.data_mut()[0] = 2;

        assert!(
            bm.flush_page(block_id).is_err(),
            "flush_page must not read a dirty frame while a mutable lease is active"
        );
        assert!(
            bm.flush_all().is_err(),
            "flush_all must not read dirty frames while a mutable lease is active"
        );

        drop(write_guard);
        bm.flush_page(block_id)
            .expect("flush succeeds after mutable lease release");
    }

    #[test]
    fn test_multiple_pages() {
        let bm = create_buffer_manager(10);
        let mut block_ids = Vec::new();

        // Create several pages
        for i in 0..5 {
            let mut guard = bm.new_page().expect("new_page");
            guard.data_mut()[0] = i as u8;
            block_ids.push(guard.block_id());
        }

        // Verify all pages
        for (i, &block_id) in block_ids.iter().enumerate() {
            let guard = bm.fetch_page(block_id).expect("fetch_page");
            assert_eq!(guard.data()[0], i as u8);
        }
    }

    #[test]
    fn test_eviction() {
        // Create a small buffer pool
        let bm = create_buffer_manager(3);
        let mut block_ids = Vec::new();

        // Create more pages than the pool can hold
        for i in 0..10 {
            let mut guard = bm.new_page().expect("new_page");
            guard.data_mut()[0] = i as u8;
            block_ids.push(guard.block_id());
        }

        // All pages should still be accessible (via eviction and reload)
        for (i, &block_id) in block_ids.iter().enumerate() {
            let guard = bm.fetch_page(block_id).expect("fetch_page");
            assert_eq!(guard.data()[0], i as u8, "Page {} corrupted", i);
        }
    }

    #[test]
    fn test_stats() {
        let bm = create_buffer_manager(10);

        let initial_stats = bm.stats();
        assert_eq!(initial_stats.total_frames, 10);
        assert_eq!(initial_stats.free_frames, 10);
        assert_eq!(initial_stats.used_frames, 0);

        // Create some pages
        let guard1 = bm.new_page().expect("new_page");
        let _guard2 = bm.new_page().expect("new_page");

        let stats = bm.stats();
        assert_eq!(stats.used_frames, 2);
        assert_eq!(stats.free_frames, 8);
        assert!(stats.pinned_frames >= 2); // Both guards are still held
        assert!(stats.dirty_frames >= 2);

        drop(guard1);

        let stats = bm.stats();
        assert!(stats.pinned_frames >= 1);
    }

    #[test]
    fn test_flush_all() {
        let bm = create_buffer_manager(10);

        // Create and modify some pages
        for i in 0..5 {
            let mut guard = bm.new_page().expect("new_page");
            guard.data_mut()[0] = i as u8;
        }

        // Flush all
        bm.flush_all().expect("flush_all");

        // Check that dirty count is 0
        let stats = bm.stats();
        assert_eq!(stats.dirty_frames, 0);
    }

    #[test]
    fn test_delete_page() {
        let bm = create_buffer_manager(10);

        // Create a page
        let guard = bm.new_page().expect("new_page");
        let block_id = guard.block_id();
        drop(guard);

        // Delete it
        bm.delete_page(block_id).expect("delete_page");

        // Stats should show one less used frame
        let stats = bm.stats();
        assert_eq!(stats.used_frames, 0);
    }

    #[test]
    fn test_pinned_page_not_evicted() {
        // Create a very small buffer pool
        let bm = create_buffer_manager(2);

        // Pin one page
        let pinned_guard = bm.new_page().expect("new_page");
        let pinned_block = pinned_guard.block_id();

        // Fill the pool with another page
        let mut other_guard = bm.new_page().expect("new_page");
        other_guard.data_mut()[0] = 99;
        drop(other_guard);

        // Try to create more pages (should evict the unpinned one)
        let _new_guard = bm.new_page().expect("new_page - should evict unpinned");

        // The pinned page should still be valid
        assert_eq!(pinned_guard.block_id(), pinned_block);
    }
}