armdb 0.7.0

use std::fs;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};

use crate::sync::{self, Mutex, MutexGuard};

use crate::disk_loc::DiskLoc;
use crate::entry::{self, make_tombstone_gsn, serialize_entry};
use crate::error::DbResult;
use crate::io::aligned_buf::AlignedBuf;
use crate::io::direct;

#[cfg(feature = "encryption")]
use crate::crypto::PageCipher;
#[cfg(feature = "encryption")]
use crate::io::tags::{self, TagFile};

#[cfg(feature = "encryption")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum EncryptedFlushMode {
    /// Only complete pages; partial page stays in buffer.
    NonForce,
    /// Pad partial to full page, write, advance base_offset past it.
    /// Used from rotate() and Shard::flush() where appends continue at
    /// the next page.
    ForceAdvance,
}

/// In-memory write buffer. Entries are accumulated here and flushed to disk
/// in batch when full, on rotation, or on explicit flush/close.
pub(crate) struct WriteBuffer {
    buf: AlignedBuf,
    len: usize,
    base_offset: u64,
}

impl WriteBuffer {
    fn new(capacity: usize, base_offset: u64) -> Self {
        Self {
            buf: AlignedBuf::zeroed(capacity),
            len: 0,
            base_offset,
        }
    }

    /// Append serialized entry data to the buffer. Returns the file offset
    /// where the data will land on disk after flush.
    fn append(&mut self, data: &[u8]) -> u64 {
        let offset = self.base_offset + self.len as u64;
        self.buf[self.len..self.len + data.len()].copy_from_slice(data);
        self.len += data.len();
        offset
    }

    /// Read bytes from the buffer by absolute file offset.
    /// Returns None if the requested range is outside the buffer.
    #[cfg(feature = "var-collections")]
    pub(crate) fn read(&self, file_offset: u64, len: usize) -> Option<&[u8]> {
        if file_offset >= self.base_offset {
            let start = (file_offset - self.base_offset) as usize;
            if start + len <= self.len {
                return Some(&self.buf[start..start + len]);
            }
        }
        None
    }

    fn is_full(&self, needed: usize) -> bool {
        self.buf.len() - self.len < needed
    }

    #[inline]
    fn capacity(&self) -> usize {
        self.buf.len()
    }

    fn data(&self) -> &[u8] {
        &self.buf[..self.len]
    }

    fn reset(&mut self, new_base: u64) {
        self.len = 0;
        self.base_offset = new_base;
    }

    /// Shift completed bytes out of the buffer, keeping the remainder.
    fn compact(&mut self, flushed_bytes: usize) {
        let remainder = self.len - flushed_bytes;
        if remainder > 0 {
            self.buf.copy_within(flushed_bytes..self.len, 0);
        }
        self.base_offset += flushed_bytes as u64;
        self.len = remainder;
    }
}

pub struct Shard {
    pub id: u8,
    dir: PathBuf,
    gsn: Arc<AtomicU64>,
    /// Count of replication entries dropped because the SPSC ring was full
    /// during streaming (F1). Bumped by `append_entry` on a failed `push`;
    /// watched lock-free by the replication server to trigger an overflow
    /// recovery catch-up round so no live entry is permanently lost. Shares the
    /// `Arc` with [`ShardInner::replication_overflow`].
    #[cfg(feature = "replication")]
    replication_overflow: Arc<AtomicU64>,
    /// Highest **durable** GSN this shard recovered from disk at open (F3). The
    /// replication follower clamps its reconnect `from_gsn` to this so a cursor
    /// fsynced past the last flushed entry (crash between cursor save and shard
    /// flush) refetches the lost tail instead of skipping it. Set once, right
    /// after recovery.
    #[cfg(feature = "replication")]
    durable_recovered_gsn: AtomicU64,
    inner: Mutex<ShardInner>,
}

pub struct ShardInner {
    pub(crate) active: ActiveFile,
    pub(crate) write_buf: WriteBuffer,
    pub(crate) immutable: Vec<std::sync::Arc<ImmutableFile>>,
    pub(crate) dead_bytes: std::collections::HashMap<u32, u64>,
    pub(crate) key_len: Option<usize>,
    pub(crate) hints: bool,
    pub(crate) next_file_id: u32,
    pub(crate) max_file_size: u64,
    /// Max immutable files one compaction pass retires for this shard
    /// (`0` = unlimited). Mirrors [`crate::Config::compaction_max_files_per_pass`];
    /// set by the engine at open. Read under the lock in `compact_shard_inner`.
    pub(crate) compaction_max_files_per_pass: usize,
    pub(crate) last_compaction_output_ids: Vec<u32>,
    pub(crate) effective_direct_io: bool,
    pub(crate) page_aligned: bool,
    gsn: Arc<AtomicU64>,
    /// See [`Shard::replication_overflow`]; same `Arc`, written here from
    /// `append_entry` when the replication ring rejects a push.
    #[cfg(feature = "replication")]
    pub(crate) replication_overflow: Arc<AtomicU64>,
    #[cfg(all(target_os = "linux", feature = "io-uring"))]
    uring_writer: Option<crate::io::uring::UringWriter>,
    #[cfg(feature = "encryption")]
    pub(crate) cipher: Option<Arc<PageCipher>>,
    #[cfg(feature = "replication")]
    pub(crate) replication_tx: Option<rtrb::Producer<crate::replication::ReplicationEntry>>,
    /// Test-only fault-injection seam: when set, `rotate()` treats hint
    /// generation for the just-retired file as if it hit a `CorruptedEntry`,
    /// exercising the early-stop path (F2) without a physically corrupt file.
    #[cfg(test)]
    pub(crate) test_force_hint_corruption: bool,
}

pub(crate) struct ActiveFile {
    pub(crate) file: std::fs::File,
    pub(crate) read_file: Arc<std::fs::File>,
    pub(crate) file_id: u32,
    pub(crate) write_offset: u64,
    pub(crate) path: PathBuf,
    #[cfg(feature = "encryption")]
    pub(crate) tag_file: Option<Arc<TagFile>>,
}

pub(crate) struct ImmutableFile {
    pub(crate) file: std::fs::File,
    pub(crate) file_id: u32,
    #[cfg(feature = "encryption")]
    pub(crate) path: PathBuf,
    pub(crate) total_bytes: u64,
    #[cfg(feature = "encryption")]
    pub(crate) tag_file: Option<Arc<TagFile>>,
}

/// The shard set of one Bitcask collection.
///
/// A distinct owner rather than a bare `Vec<Shard>` because the directory lock
/// lease lives here (Task 4 of the file-lock plan): the replication threads
/// clone this `Arc` out of the `Engine` and outlive it, so anything able to
/// reach the shard files must hold the lease along with them. Putting the lease
/// on `Engine` instead would let it drop while those threads still write.
///
/// See `docs/superpowers/specs/26-07-15-db-directory-file-lock.md`.
#[derive(derive_more::Deref)]
pub struct Shards {
    #[deref]
    shards: Vec<Shard>,
    /// Exclusive lease on the collection directory.
    ///
    /// Declared last: fields drop in declaration order, so the shards flush and
    /// close before the lock is released. Releasing first would let the next
    /// process open the directory mid-flush.
    _lock: fs::File,
}

impl Shards {
    pub(crate) fn new(shards: Vec<Shard>, lock: fs::File) -> Self {
        Self {
            shards,
            _lock: lock,
        }
    }
}

impl Shard {
    /// Open or create a shard in the given directory.
    #[allow(clippy::too_many_arguments)]
    pub fn open(
        id: u8,
        dir: &Path,
        max_file_size: u64,
        write_buffer_size: usize,
        hints: bool,
        direct_io: bool,
        effective_direct_io: bool,
        io_backend: crate::config::IoBackend,
        gsn: Arc<AtomicU64>,
    ) -> DbResult<Self> {
        Self::open_inner(
            id,
            dir,
            max_file_size,
            write_buffer_size,
            hints,
            direct_io,
            effective_direct_io,
            io_backend,
            #[cfg(feature = "encryption")]
            None,
            gsn,
        )
    }

    /// Open or create a shard with optional encryption.
    #[cfg(feature = "encryption")]
    #[allow(clippy::too_many_arguments)]
    pub fn open_encrypted(
        id: u8,
        dir: &Path,
        max_file_size: u64,
        write_buffer_size: usize,
        hints: bool,
        direct_io: bool,
        effective_direct_io: bool,
        io_backend: crate::config::IoBackend,
        cipher: Option<Arc<PageCipher>>,
        gsn: Arc<AtomicU64>,
    ) -> DbResult<Self> {
        Self::open_inner(
            id,
            dir,
            max_file_size,
            write_buffer_size,
            hints,
            direct_io,
            effective_direct_io,
            io_backend,
            cipher,
            gsn,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn open_inner(
        id: u8,
        dir: &Path,
        max_file_size: u64,
        write_buffer_size: usize,
        hints: bool,
        direct_io: bool,
        effective_direct_io: bool,
        #[cfg_attr(
            not(all(target_os = "linux", feature = "io-uring")),
            allow(unused_variables)
        )]
        io_backend: crate::config::IoBackend,
        #[cfg(feature = "encryption")] cipher: Option<Arc<PageCipher>>,
        gsn: Arc<AtomicU64>,
    ) -> DbResult<Self> {
        fs::create_dir_all(dir)?;

        // F1: shared per-shard replication overflow counter — one `Arc` held by
        // both `Shard` (lock-free reads from the replication server) and
        // `ShardInner` (bumped from `append_entry` when a ring push fails).
        #[cfg(feature = "replication")]
        let replication_overflow = Arc::new(AtomicU64::new(0));

        // Scan existing data files
        let mut file_ids: Vec<u32> = Vec::new();
        for entry in fs::read_dir(dir)? {
            let entry = entry?;
            let name = entry.file_name();
            let name = name.to_string_lossy();
            if name.ends_with(".data")
                && let Ok(id) = name.trim_end_matches(".data").parse::<u32>()
            {
                file_ids.push(id);
            }
        }
        file_ids.sort();

        // Sweep crash-leftover temp files (V-3/V-9 fix).
        for entry in fs::read_dir(dir)? {
            let entry = entry?;
            let name = entry.file_name();
            let name = name.to_string_lossy();
            if name.ends_with(".data.tmp")
                || name.ends_with(".tags.tmp")
                || name.ends_with(".hint.tmp")
            {
                match fs::remove_file(entry.path()) {
                    Ok(()) => {}
                    Err(e) if e.kind() == std::io::ErrorKind::NotFound => {}
                    Err(e) => return Err(e.into()),
                }
            }
        }

        let mut immutable = Vec::new();

        #[cfg(feature = "encryption")]
        let has_cipher = cipher.is_some();

        let page_aligned = {
            #[cfg(feature = "encryption")]
            {
                direct_io || cipher.is_some()
            }
            #[cfg(not(feature = "encryption"))]
            {
                direct_io
            }
        };

        if file_ids.is_empty() {
            // Create first data file
            let file_id = 1u32;
            let path = dir.join(format!("{file_id:06}.data"));
            let file = direct::open_serving_write(&path, effective_direct_io)?;
            let read_file = Arc::new(direct::open_serving_read(&path, effective_direct_io)?);

            #[cfg(feature = "encryption")]
            let tag_file = if has_cipher {
                Some(Arc::new(TagFile::open_write(&tags::tags_path_for_data(
                    &path,
                ))?))
            } else {
                None
            };

            // F5: fsync the shard directory so the freshly created data (and tag)
            // file's directory entry is durable across a crash.
            direct::sync_dir(dir)?;

            let active = ActiveFile {
                file,
                read_file,
                file_id,
                write_offset: 0,
                path,
                #[cfg(feature = "encryption")]
                tag_file,
            };

            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            let uring_writer = match io_backend {
                crate::config::IoBackend::Uring { sqpoll_idle_ms } => {
                    use std::os::unix::io::AsRawFd;
                    let mut w = crate::io::uring::UringWriter::new(sqpoll_idle_ms)?;
                    w.set_file(active.file.as_raw_fd());
                    w.set_direct(effective_direct_io);
                    Some(w)
                }
                crate::config::IoBackend::Pwrite => None,
            };

            return Ok(Self {
                id,
                dir: dir.to_path_buf(),
                gsn: gsn.clone(),
                #[cfg(feature = "replication")]
                replication_overflow: replication_overflow.clone(),
                #[cfg(feature = "replication")]
                durable_recovered_gsn: AtomicU64::new(0),
                inner: Mutex::new(ShardInner {
                    active,
                    write_buf: WriteBuffer::new(write_buffer_size, 0),
                    immutable,
                    dead_bytes: std::collections::HashMap::new(),
                    key_len: None,
                    hints,
                    next_file_id: 2,
                    max_file_size,
                    compaction_max_files_per_pass:
                        crate::config::DEFAULT_COMPACTION_MAX_FILES_PER_PASS,
                    last_compaction_output_ids: Vec::new(),
                    effective_direct_io,
                    page_aligned,
                    gsn,
                    #[cfg(feature = "replication")]
                    replication_overflow: replication_overflow.clone(),
                    #[cfg(all(target_os = "linux", feature = "io-uring"))]
                    uring_writer,
                    #[cfg(feature = "encryption")]
                    cipher,
                    #[cfg(feature = "replication")]
                    replication_tx: None,
                    #[cfg(test)]
                    test_force_hint_corruption: false,
                }),
            });
        }

        // Last file is active, rest are immutable
        let active_id = file_ids.pop().expect("file_ids is not empty");
        for &fid in &file_ids {
            let path = dir.join(format!("{fid:06}.data"));
            let file = direct::open_serving_read(&path, effective_direct_io)?;
            let total_bytes = file.metadata()?.len();

            #[cfg(feature = "encryption")]
            let tag_file = if has_cipher {
                let tp = tags::tags_path_for_data(&path);
                if tp.exists() {
                    Some(Arc::new(TagFile::open_read(&tp)?))
                } else {
                    None
                }
            } else {
                None
            };

            immutable.push(std::sync::Arc::new(ImmutableFile {
                file,
                file_id: fid,
                #[cfg(feature = "encryption")]
                path,
                total_bytes,
                #[cfg(feature = "encryption")]
                tag_file,
            }));
        }

        let active_path = dir.join(format!("{active_id:06}.data"));
        let active_file = direct::open_serving_write(&active_path, effective_direct_io)?;
        let write_offset = active_file.metadata()?.len();

        let active_read = Arc::new(direct::open_serving_read(
            &active_path,
            effective_direct_io,
        )?);

        #[cfg(feature = "encryption")]
        let tag_file = if has_cipher {
            Some(Arc::new(TagFile::open_write(&tags::tags_path_for_data(
                &active_path,
            ))?))
        } else {
            None
        };

        let active = ActiveFile {
            file: active_file,
            read_file: active_read,
            file_id: active_id,
            write_offset,
            path: active_path,
            #[cfg(feature = "encryption")]
            tag_file,
        };

        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        let uring_writer = match io_backend {
            crate::config::IoBackend::Uring { sqpoll_idle_ms } => {
                use std::os::unix::io::AsRawFd;
                let mut w = crate::io::uring::UringWriter::new(sqpoll_idle_ms)?;
                w.set_file(active.file.as_raw_fd());
                w.set_direct(effective_direct_io);
                Some(w)
            }
            crate::config::IoBackend::Pwrite => None,
        };

        Ok(Self {
            id,
            dir: dir.to_path_buf(),
            gsn: gsn.clone(),
            #[cfg(feature = "replication")]
            replication_overflow: replication_overflow.clone(),
            #[cfg(feature = "replication")]
            durable_recovered_gsn: AtomicU64::new(0),
            inner: Mutex::new(ShardInner {
                active,
                write_buf: WriteBuffer::new(write_buffer_size, write_offset),
                immutable,
                dead_bytes: std::collections::HashMap::new(),
                key_len: None,
                hints,
                next_file_id: active_id
                    .checked_add(1)
                    .ok_or(crate::error::DbError::Client(
                        "file_id space exhausted at open",
                    ))?,
                max_file_size,
                compaction_max_files_per_pass: crate::config::DEFAULT_COMPACTION_MAX_FILES_PER_PASS,
                last_compaction_output_ids: Vec::new(),
                effective_direct_io,
                page_aligned,
                gsn,
                #[cfg(feature = "replication")]
                replication_overflow: replication_overflow.clone(),
                #[cfg(all(target_os = "linux", feature = "io-uring"))]
                uring_writer,
                #[cfg(feature = "encryption")]
                cipher,
                #[cfg(feature = "replication")]
                replication_tx: None,
                #[cfg(test)]
                test_force_hint_corruption: false,
            }),
        })
    }

    pub(crate) fn effective_direct_io(&self) -> bool {
        self.lock().effective_direct_io
    }

    /// Per-collection sequence counter (lock-free read).
    pub fn gsn(&self) -> &AtomicU64 {
        &self.gsn
    }

    /// Lock the shard for writing. The caller holds the lock, performs disk write +
    /// index update atomically, then drops the guard.
    pub fn lock(&self) -> MutexGuard<'_, ShardInner> {
        sync::lock(&self.inner)
    }

    #[cfg(feature = "var-collections")]
    /// Read a 4096-byte aligned block from a data file.
    /// Returns `(block, is_full_block)` — `is_full_block` is true only if the
    /// block is entirely within the file's data region. Partial blocks (at the
    /// end of a file, padded with zeros) must NOT be cached because subsequent
    /// reads would return stale zero tails.
    pub fn read_block(
        &self,
        file_id: u32,
        block_offset: u64,
    ) -> DbResult<(crate::io::aligned_buf::AlignedBuf, bool)> {
        let refs = {
            let inner = sync::lock(&self.inner);
            inner.resolve_block_refs(file_id)?
        };
        pread_decrypt_block(file_id, block_offset, &refs)
    }

    /// Read a whole value at `(file_id, offset)` of length `len`, without holding
    /// the shard write-lock for the disk I/O. Mirror of [`Shard::read_block`]:
    /// briefly locks to resolve file refs, then `pread`s outside the lock.
    /// Returns plaintext when encryption is enabled. `StaleDiskLoc` if the file
    /// was compacted away.
    #[cfg(feature = "var-collections")]
    pub fn read_value(&self, file_id: u32, offset: u32, len: usize) -> DbResult<Vec<u8>> {
        let refs = {
            let inner = sync::lock(&self.inner);
            inner.resolve_block_refs(file_id)?
        };
        let file: &std::fs::File = match (&refs.active_read, &refs.immutable_file) {
            (Some(f), _) => f,
            (None, Some(arc)) => &arc.file,
            (None, None) => return Err(crate::error::DbError::StaleDiskLoc),
        };
        #[cfg(feature = "encryption")]
        if let Some(cipher) = &refs.cipher {
            let tag_file = refs
                .tag_file
                .as_ref()
                .ok_or(crate::error::DbError::StaleDiskLoc)?;
            return crate::io::direct::pread_value_encrypted(
                file,
                tag_file,
                cipher,
                file_id,
                offset as u64,
                len,
            );
        }
        crate::io::direct::pread_value(file, offset as u64, len)
    }

    /// Get shard directory for recovery.
    pub fn dir(&self) -> &Path {
        &self.dir
    }

    /// Get all file IDs in order (immutable + active).
    pub fn file_ids(&self) -> Vec<u32> {
        let inner = sync::lock(&self.inner);
        let mut ids: Vec<u32> = inner.immutable.iter().map(|f| f.file_id).collect();
        ids.push(inner.active.file_id);
        ids
    }

    /// Get all (file_id, total_bytes) pairs for cache warmup.
    pub fn file_sizes(&self) -> Vec<(u32, u64)> {
        let inner = sync::lock(&self.inner);
        let mut result: Vec<(u32, u64)> = inner
            .immutable
            .iter()
            .map(|f| (f.file_id, f.total_bytes))
            .collect();
        result.push((inner.active.file_id, inner.active.write_offset));
        result
    }

    /// Generate a hint file for the current active data file.
    /// Called during graceful shutdown.
    pub fn write_active_hint(&self, key_len: usize) -> DbResult<()> {
        let mut inner = sync::lock(&self.inner);
        if inner.active.write_offset == 0 {
            return Ok(()); // empty active file
        }
        // flush_write_buf_final pads partial trailing page for encryption
        inner.flush_write_buf_final()?;
        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        match &mut inner.uring_writer {
            Some(w) => w.fsync()?,
            None => direct::fsync(&inner.active.file)?,
        }
        #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
        direct::fsync(&inner.active.file)?;

        // Sync tag file before generating hints so tags are readable during recovery
        #[cfg(feature = "encryption")]
        if let Some(ref tag_file) = inner.active.tag_file {
            tag_file.sync()?;
        }

        let read_file = direct::open_serving_read(&inner.active.path, inner.effective_direct_io)?;
        // write_offset tracks the logical data end (set during each append, before any
        // padding flush). Using it instead of metadata().len() prevents the encrypted
        // final-page padding region from appearing as a phantom hint entry.
        let file_len = inner.active.write_offset;

        #[cfg(feature = "encryption")]
        let hint_data = if let (Some(cipher), Some(tag_file)) =
            (&inner.cipher, &inner.active.tag_file)
        {
            match crate::hint::generate_hint_data_dyn_encrypted(
                &read_file,
                file_len,
                key_len,
                cipher,
                tag_file.as_ref(),
                inner.active.file_id,
            ) {
                Ok(data) => data,
                Err(crate::error::DbError::CorruptedEntry { offset }) => {
                    tracing::warn!(offset, "hint generation stopped early — skipping hint file");
                    return Ok(());
                }
                Err(e) => return Err(e),
            }
        } else {
            match crate::hint::generate_hint_data_dyn(&read_file, file_len, key_len) {
                Ok(data) => data,
                Err(crate::error::DbError::CorruptedEntry { offset }) => {
                    tracing::warn!(offset, "hint generation stopped early — skipping hint file");
                    return Ok(());
                }
                Err(e) => return Err(e),
            }
        };
        #[cfg(not(feature = "encryption"))]
        let hint_data = match crate::hint::generate_hint_data_dyn(&read_file, file_len, key_len) {
            Ok(data) => data,
            Err(crate::error::DbError::CorruptedEntry { offset }) => {
                tracing::warn!(offset, "hint generation stopped early — skipping hint file");
                return Ok(());
            }
            Err(e) => return Err(e),
        };

        let hint_path = crate::hint::hint_path_for_data(&inner.active.path);
        crate::hint::write_hint_file(&hint_path, &hint_data)?;
        Ok(())
    }

    /// Flush the write buffer to disk (without fsync).
    pub fn flush_buf(&self) -> DbResult<()> {
        let mut inner = sync::lock(&self.inner);
        inner.flush_write_buf()
    }

    /// Force-flush the write buffer for replication catch-up.
    ///
    /// In plain mode behaves like `flush_buf`. In encrypted mode pads and
    /// encrypts the trailing partial page (ForceAdvance) so the encrypted
    /// `ShardLogReader` can decrypt every entry whose GSN is below
    /// `shard.gsn().load()`. The padding gap is acceptable — it becomes
    /// dead space at the end of the active file.
    pub fn flush_for_replication_catchup(&self) -> DbResult<()> {
        let mut inner = sync::lock(&self.inner);
        inner.flush_write_buf_final()
    }

    /// Flush write buffer + fsync the active file and tag file to disk.
    pub fn flush(&self) -> DbResult<()> {
        let mut inner = sync::lock(&self.inner);
        inner.flush_write_buf_final()?;
        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        match &mut inner.uring_writer {
            Some(w) => w.fsync()?,
            None => direct::fsync(&inner.active.file)?,
        }
        #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
        direct::fsync(&inner.active.file)?;
        #[cfg(feature = "encryption")]
        if let Some(ref tag_file) = inner.active.tag_file {
            tag_file.sync()?;
        }
        Ok(())
    }
}

impl Shard {
    #[cfg(feature = "replication")]
    pub(crate) fn catchup_snapshot(
        &self,
    ) -> DbResult<crate::replication::snapshot::ShardCatchupSnapshot> {
        use crate::replication::snapshot::{CatchupFile, ShardCatchupSnapshot};

        let mut inner = sync::lock(&self.inner);
        inner.flush_write_buf_final()?;
        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        match &mut inner.uring_writer {
            Some(writer) => writer.fsync()?,
            None => direct::fsync(&inner.active.file)?,
        }
        #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
        direct::fsync(&inner.active.file)?;
        #[cfg(feature = "encryption")]
        if let Some(tag_file) = &inner.active.tag_file {
            tag_file.sync()?;
        }

        let max_gsn = self.gsn.load(Ordering::Relaxed).saturating_sub(1);
        let overflow_baseline = self.replication_overflow.load(Ordering::Relaxed);
        let mut files = Vec::with_capacity(inner.immutable.len() + 1);
        files.extend(inner.immutable.iter().cloned().map(|file| {
            CatchupFile::immutable(
                file,
                #[cfg(feature = "encryption")]
                inner.cipher.clone(),
            )
        }));
        files.push(CatchupFile::active(
            inner.active.file_id,
            inner.active.write_offset,
            inner.active.read_file.clone(),
            #[cfg(feature = "encryption")]
            inner.active.tag_file.clone(),
            #[cfg(feature = "encryption")]
            inner.cipher.clone(),
        ));
        Ok(ShardCatchupSnapshot::new(
            self.id,
            max_gsn,
            overflow_baseline,
            files,
        ))
    }

    /// Get the active file ID (for replication index updates).
    pub fn active_file_id(&self) -> u32 {
        sync::lock(&self.inner).active.file_id
    }

    /// Return a clone of the encryption cipher, if encryption is enabled for this shard.
    /// Plumbed into `ShardLogReader` for replication catch-up.
    #[cfg(feature = "encryption")]
    pub fn cipher(&self) -> Option<Arc<crate::crypto::PageCipher>> {
        sync::lock(&self.inner).cipher.clone()
    }

    /// Install a replication SPSC producer into this shard.
    #[cfg(feature = "replication")]
    pub fn set_replication_producer(
        &self,
        producer: rtrb::Producer<crate::replication::ReplicationEntry>,
    ) {
        sync::lock(&self.inner).replication_tx = Some(producer);
    }

    /// Number of replication entries dropped so far because the SPSC ring was
    /// full (F1). Monotonic; a change since a prior read means the stream lost
    /// entries and the server must run an overflow recovery catch-up round.
    /// Lock-free.
    #[cfg(feature = "replication")]
    pub fn replication_overflow(&self) -> u64 {
        self.replication_overflow.load(Ordering::Relaxed)
    }

    /// Record the highest durable GSN recovered for this shard at open (F3).
    /// Called once by the collection open path after recovery.
    #[cfg(feature = "replication")]
    pub fn set_durable_recovered_gsn(&self, gsn: u64) {
        self.durable_recovered_gsn.store(gsn, Ordering::Relaxed);
    }

    /// Highest durable GSN recovered for this shard at open (F3). The
    /// replication follower clamps its reconnect `from_gsn` to this. Lock-free.
    #[cfg(feature = "replication")]
    pub fn durable_recovered_gsn(&self) -> u64 {
        self.durable_recovered_gsn.load(Ordering::Relaxed)
    }
}

impl Shard {
    /// Store key length so that `Drop` can write hint files automatically.
    /// Called once during tree open, after recovery.
    pub(crate) fn set_key_len(&self, key_len: usize) {
        sync::lock(&self.inner).key_len = Some(key_len);
    }

    /// Apply a [`crate::recovery::ActiveTail`] to this shard's active file.
    ///
    /// Plain (non-page-aligned) mode: truncates the active data file to
    /// `last_valid_offset` and resets the write buffer/offset so subsequent
    /// appends overwrite any post-crash garbage tail.
    ///
    /// Page-aligned mode (`direct_io` or encryption): does NOT truncate
    /// (truncation would break per-page alignment / tag-file layout). Instead,
    /// advances the write buffer and `write_offset` to the next 4096-byte page
    /// boundary past `last_valid_offset`, abandoning trailing padded bytes as
    /// dead space.
    pub(crate) fn apply_recovery_tail(&self, tail: &crate::recovery::ActiveTail) -> DbResult<()> {
        let mut inner = sync::lock(&self.inner);
        if inner.active.file_id != tail.file_id {
            return Err(crate::error::DbError::Client(
                "apply_recovery_tail: file_id mismatch",
            ));
        }

        if inner.page_aligned {
            const PAGE_SIZE: u64 = 4096;
            #[cfg_attr(not(feature = "encryption"), allow(unused_mut))]
            let mut sealed_pages = tail.last_valid_offset.div_ceil(PAGE_SIZE);
            // F6: an encrypted crash can leave ciphertext pages on `.data` (and
            // tags on `.tags`) past `last_valid_offset`. A post-recovery append
            // must never re-seal a page already encrypted under this
            // (subkey, nonce) — advance the base past every page either file
            // proves was touched.
            #[cfg(feature = "encryption")]
            if inner.cipher.is_some() {
                let data_len = inner.active.file.metadata()?.len();
                sealed_pages = sealed_pages.max(data_len.div_ceil(PAGE_SIZE));
                if let Some(tag_file) = inner.active.tag_file.as_ref() {
                    let tag_len = tag_file.byte_len()?;
                    sealed_pages =
                        sealed_pages.max(tag_len.div_ceil(crate::crypto::TAG_LEN as u64));
                }
            }
            let new_base = sealed_pages * PAGE_SIZE;
            inner.write_buf.reset(new_base);
            inner.active.write_offset = new_base;
            return Ok(());
        }

        inner.active.file.set_len(tail.last_valid_offset)?;
        inner.write_buf.reset(tail.last_valid_offset);
        inner.active.write_offset = tail.last_valid_offset;
        Ok(())
    }

    /// Replace the shard's dead_bytes map with the one computed during recovery.
    pub(crate) fn install_dead_bytes(&self, dead_bytes: std::collections::HashMap<u32, u64>) {
        let mut inner = sync::lock(&self.inner);
        inner.dead_bytes = dead_bytes;
    }

    /// Set the per-pass compaction file cap (`0` = unlimited). Called by the
    /// engine at open to propagate [`crate::Config::compaction_max_files_per_pass`].
    pub(crate) fn set_compaction_max_files_per_pass(&self, cap: usize) {
        sync::lock(&self.inner).compaction_max_files_per_pass = cap;
    }
}

impl Drop for Shard {
    fn drop(&mut self) {
        let inner = sync::lock(&self.inner);
        let key_len = inner.key_len;
        let hints = inner.hints;
        drop(inner);
        if hints && let Some(kl) = key_len {
            if let Err(e) = self.write_active_hint(kl) {
                tracing::error!(shard_id = self.id, "failed to write hint on drop: {e}");
                // Hint writing includes flushing, so if it failed we must
                // still attempt a plain flush to avoid data loss.
                if let Err(e2) = self.flush() {
                    tracing::error!(shard_id = self.id, "fallback flush also failed: {e2}");
                }
            }
            return;
        }
        if let Err(e) = self.flush() {
            tracing::error!(shard_id = self.id, "failed to flush shard on drop: {e}");
        }
    }
}

impl ShardInner {
    /// Flush the in-memory write buffer to disk.
    /// When encryption is enabled, only complete 4096-byte pages are flushed;
    /// the partial trailing page remains in the buffer.
    pub(crate) fn flush_write_buf(&mut self) -> DbResult<()> {
        if self.write_buf.len == 0 {
            return Ok(());
        }

        #[cfg(feature = "encryption")]
        if self.cipher.is_some() {
            return self.flush_write_buf_encrypted(EncryptedFlushMode::NonForce);
        }

        self.flush_write_buf_plain(false)
    }

    /// Flush all data including partial trailing page (pad to 4096).
    /// Used before rotation and on close.
    pub(crate) fn flush_write_buf_final(&mut self) -> DbResult<()> {
        if self.write_buf.len == 0 {
            return Ok(());
        }

        #[cfg(feature = "encryption")]
        if self.cipher.is_some() {
            return self.flush_write_buf_encrypted(EncryptedFlushMode::ForceAdvance);
        }

        self.flush_write_buf_plain(true)
    }

    fn flush_write_buf_plain(&mut self, force: bool) -> DbResult<()> {
        let data_len = self.write_buf.len;
        if data_len == 0 {
            return Ok(());
        }

        // Legacy byte-granular path: flush everything.
        if !self.page_aligned {
            let data = self.write_buf.data();
            let offset = self.write_buf.base_offset;
            #[cfg(all(target_os = "linux", feature = "io-uring"))]
            match &mut self.uring_writer {
                Some(w) => w.write_at(data, offset)?,
                None => direct::pwrite_at(&self.active.file, data, offset)?,
            }
            #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
            direct::pwrite_at(&self.active.file, data, offset)?;
            let flushed = data.len() as u64;
            self.write_buf.reset(offset + flushed);
            metrics::counter!("armdb.flush.count").increment(1);
            metrics::counter!("armdb.flush.bytes").increment(flushed);
            return Ok(());
        }

        // Page-granular path (mirrors flush_write_buf_encrypted, no encryption).
        let complete_bytes = (data_len / 4096) * 4096;
        let remainder = data_len % 4096;
        let base_offset = self.write_buf.base_offset;

        let (flush_bytes, original_len) = if force && remainder > 0 {
            let target = complete_bytes + 4096;
            let original = self.write_buf.len;
            for i in original..target {
                self.write_buf.buf[i] = 0;
            }
            self.write_buf.len = target;
            (target, Some(original))
        } else {
            if complete_bytes == 0 {
                return Ok(());
            }
            (complete_bytes, None)
        };

        let data = &self.write_buf.buf[..flush_bytes];
        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        let res = match &mut self.uring_writer {
            Some(w) => w.write_at(data, base_offset),
            None => direct::pwrite_at(&self.active.file, data, base_offset),
        };
        #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
        let res = direct::pwrite_at(&self.active.file, data, base_offset);
        if let Err(e) = res {
            if let Some(orig) = original_len {
                for i in orig..self.write_buf.len {
                    self.write_buf.buf[i] = 0;
                }
                self.write_buf.len = orig;
            }
            return Err(e);
        }

        metrics::counter!("armdb.flush.count").increment(1);
        metrics::counter!("armdb.flush.bytes").increment(flush_bytes as u64);

        if force {
            self.write_buf.reset(base_offset + flush_bytes as u64);
        } else {
            self.write_buf.compact(complete_bytes);
        }
        Ok(())
    }

    #[cfg(feature = "encryption")]
    fn flush_write_buf_encrypted(&mut self, mode: EncryptedFlushMode) -> DbResult<()> {
        let cipher = self
            .cipher
            .as_ref()
            .expect("caller checked cipher.is_some()");
        let data_len = self.write_buf.len;
        let complete_bytes = (data_len / 4096) * 4096;
        let remainder = data_len % 4096;

        let force = mode == EncryptedFlushMode::ForceAdvance;

        let (flush_bytes, original_len) = if force && remainder > 0 {
            let target = complete_bytes + 4096;
            let original = self.write_buf.len;
            for i in original..target {
                self.write_buf.buf[i] = 0;
            }
            self.write_buf.len = target;
            (target, Some(original))
        } else {
            if complete_bytes == 0 {
                return Ok(());
            }
            (complete_bytes, None)
        };

        let num_pages = flush_bytes / 4096;
        let base_offset = self.write_buf.base_offset;
        let start_page = base_offset / 4096;
        let file_id = self.active.file_id;

        let mut encrypted = AlignedBuf::zeroed(flush_bytes);
        encrypted.copy_from_slice(&self.write_buf.buf[..flush_bytes]);
        let mut tag_list = Vec::with_capacity(num_pages);
        for i in 0..num_pages {
            let page_start = i * 4096;
            let page = &mut encrypted[page_start..page_start + 4096];
            let tag = cipher.encrypt_page(file_id, start_page + i as u64, page)?;
            tag_list.push(tag);
        }

        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        let res = match &mut self.uring_writer {
            Some(w) => w.write_at(&encrypted, base_offset),
            None => direct::pwrite_at(&self.active.file, &encrypted, base_offset),
        };
        #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
        let res = direct::pwrite_at(&self.active.file, &encrypted, base_offset);
        if let Err(e) = res {
            if let Some(orig) = original_len {
                for i in orig..self.write_buf.len {
                    self.write_buf.buf[i] = 0;
                }
                self.write_buf.len = orig;
            }
            return Err(e);
        }

        if let Some(ref tag_file) = self.active.tag_file
            && let Err(e) = tag_file.write_tags(start_page, &tag_list)
        {
            if let Some(orig) = original_len {
                for i in orig..self.write_buf.len {
                    self.write_buf.buf[i] = 0;
                }
                self.write_buf.len = orig;
            }
            return Err(e);
        }

        match mode {
            EncryptedFlushMode::NonForce => {
                self.write_buf.compact(complete_bytes);
            }
            EncryptedFlushMode::ForceAdvance => {
                let new_base = base_offset + flush_bytes as u64;
                self.write_buf.reset(new_base);
            }
        }

        metrics::counter!("armdb.flush.count").increment(1);
        metrics::counter!("armdb.flush.bytes").increment(flush_bytes as u64);
        Ok(())
    }

    /// Append an entry to the write buffer. Returns (DiskLoc pointing to value, gsn).
    /// Data is NOT written to disk immediately — it stays in the buffer until flush.
    pub fn append_entry(
        &mut self,
        shard_id: u8,
        key: &[u8],
        value: &[u8],
        tombstone: bool,
    ) -> DbResult<(DiskLoc, u64)> {
        let needed = crate::entry::entry_size(key.len(), value.len() as u32) as usize;
        if needed > self.write_buf.capacity() {
            return Err(crate::error::DbError::Client(
                "entry exceeds write_buffer_size",
            ));
        }

        // V-04/F4: Rotate-before-append so value_offset fits in u32. Use the real
        // buffered write position (`base_offset + len`) — after a force-flush pads
        // a partial page, `base_offset` advances to the next page while the logical
        // `active.write_offset` lags behind by up to a page. Checking the stale
        // logical offset would under-count the append position and let
        // `value_offset` silently overflow `u32` (aliased DiskLoc). This is equal
        // to the old check except after a padding advance, where it is strictly
        // more conservative.
        let next_offset = self.write_buf.base_offset + self.write_buf.len as u64;
        if next_offset + (needed as u64) > self.max_file_size {
            self.rotate(shard_id, key.len())?;
        }

        let gsn = self.gsn.fetch_add(1, Ordering::Relaxed);
        debug_assert!(
            gsn & crate::entry::FLAGS_MASK == 0,
            "GSN overflowed into reserved flag bits"
        );
        crate::entry::guard_gsn(gsn)?;

        let buf = serialize_entry(gsn, key, value, tombstone);

        // Flush if buffer can't fit this entry
        if self.write_buf.is_full(buf.len()) {
            self.flush_write_buf()?;
            if self.write_buf.is_full(buf.len()) {
                // Encrypted shard retained a partial page that still blocks this
                // append after the non-force flush evicted complete pages. Force
                // a rotation to free buffer capacity (V-03).
                self.rotate(shard_id, key.len())?;
            }
        }

        // The entry lands at the current buffered end; `write_buf.append` returns
        // this same value. Compute it (and validate) BEFORE mutating the buffer so
        // the u32 guard rejects an over-large offset without leaving a stray entry.
        let entry_offset = self.write_buf.base_offset + self.write_buf.len as u64;
        // DiskLoc.offset points to the value data (after header + key)
        let header_and_key = size_of::<entry::EntryHeader>() + key.len();
        let value_offset = entry_offset + header_and_key as u64;
        // F4: promote the former debug_assert to a hard guard — a release build must
        // never silently truncate `value_offset` via `as u32` into an aliased
        // DiskLoc. With `max_file_size <= u32::MAX` and the corrected rotate check
        // above this is unreachable in practice; it is a defense-in-depth backstop.
        if value_offset > u32::MAX as u64 {
            return Err(crate::error::DbError::Client(
                "value_offset exceeds u32::MAX after rotation check",
            ));
        }
        let loc = DiskLoc::new(self.active.file_id, value_offset as u32, value.len() as u32);

        // memcpy only — no disk I/O
        let appended_at = self.write_buf.append(&buf);
        debug_assert_eq!(appended_at, entry_offset);

        // Push to replication SPSC channel (Vec moved, zero extra allocation)
        #[cfg(feature = "replication")]
        if let Some(tx) = &mut self.replication_tx
            && tx
                .push(crate::replication::ReplicationEntry {
                    data: buf,
                    key_len: key.len() as u16,
                })
                .is_err()
        {
            // Ring full: this entry was dropped from the stream. Its bytes are
            // in the write buffer (flushed to disk), so an overflow recovery
            // catch-up round can pick it up — but only if the server *notices*.
            // Bump the shared counter so the replication server can detect the
            // drop and re-catch-up before advancing the follower's watermark
            // past the hole (F1).
            self.replication_overflow.fetch_add(1, Ordering::Relaxed);
        }

        self.active.write_offset = self.write_buf.base_offset + self.write_buf.len as u64;

        let actual_gsn = if tombstone {
            make_tombstone_gsn(gsn)
        } else {
            gsn
        };
        Ok((loc, actual_gsn))
    }

    fn rotate(&mut self, _shard_id: u8, key_len: usize) -> DbResult<()> {
        metrics::counter!("armdb.rotation").increment(1);
        tracing::debug!(shard_id = _shard_id, "shard file rotation");
        // Capture logical end before flush_write_buf_final may pad the buffer.
        let logical_end = self.write_buf.base_offset + self.write_buf.len as u64;
        // Flush write buffer before rotation (force = pad partial page for encryption)
        self.flush_write_buf_final()?;

        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        match &mut self.uring_writer {
            Some(w) => w.fsync()?,
            None => direct::fsync(&self.active.file)?,
        }
        #[cfg(not(all(target_os = "linux", feature = "io-uring")))]
        direct::fsync(&self.active.file)?;

        // F2: every fallible operation that touches the retiring file happens
        // BEFORE the active-file swap, so a failure here leaves that file still
        // reachable as the active file — never orphaned (neither active nor a
        // member of `immutable`). The swap and the push into `immutable` are then
        // performed back-to-back with no fallible step between them.
        let old_file_id = self.active.file_id;
        let old_path = self.active.path.clone();
        let file_len = logical_end;

        // Sync the retiring tag file before reopening it read-only (its tags must
        // be on disk before hint generation reads them). Still the active file.
        #[cfg(feature = "encryption")]
        if let Some(ref tf) = self.active.tag_file {
            tf.sync()?;
        }

        // Reopen the retiring file read-only for immutable storage + hint gen.
        let imm_file = direct::open_serving_read(&old_path, self.effective_direct_io)?;

        // Open its tag file for reading (immutable storage + hint generation).
        #[cfg(feature = "encryption")]
        let imm_tag_file = if self.cipher.is_some() {
            let tp = tags::tags_path_for_data(&old_path);
            if tp.exists() {
                Some(Arc::new(TagFile::open_read(&tp)?))
            } else {
                None
            }
        } else {
            None
        };

        // Allocate + create the new active file. The retiring file is still the
        // active file, so a failure here does not orphan it.
        let new_file_id = self.allocate_file_id()?;
        let dir = old_path.parent().expect("active file has parent dir");
        let new_path = dir.join(format!("{new_file_id:06}.data"));

        // P1-7: create the new `.tags` file BEFORE its `.data`, mirroring the
        // C-03 ordering used in compaction finalization. Recovery hard-fails on a
        // `.data` that has no sibling `.tags` (an encrypted file it cannot
        // decrypt — see `recovery::make_reader`), so a fault while creating these
        // two files must never leave that shape. With tags-first, a failure in the
        // `.data` creation below leaves at worst an inert orphan `.tags` (no
        // `.data` references it and recovery never scans a lone tag file).
        #[cfg(feature = "encryption")]
        let new_tag_file = if self.cipher.is_some() {
            Some(Arc::new(TagFile::open_write(&tags::tags_path_for_data(
                &new_path,
            ))?))
        } else {
            None
        };

        let new_file = direct::open_serving_write(&new_path, self.effective_direct_io)?;
        let new_read = Arc::new(direct::open_serving_read(
            &new_path,
            self.effective_direct_io,
        )?);

        // F5: fsync the shard directory so the new data (and tag) file's directory
        // entry is durable. POSIX does not guarantee a newly created file survives
        // a crash until its containing directory is fsynced — without this the file
        // (and all its fsynced content) can vanish after reboot on some filesystems.
        direct::sync_dir(dir)?;

        // === Swap. From here until the retired file is pushed into `immutable`
        // every step is infallible, so the file is never left orphaned. ===
        let old_file = std::mem::replace(&mut self.active.file, new_file);
        self.active.path = new_path;
        self.active.file_id = new_file_id;
        self.active.write_offset = 0;
        self.active.read_file = new_read;
        self.write_buf.reset(0);

        #[cfg(feature = "encryption")]
        let old_tag_file = std::mem::replace(&mut self.active.tag_file, new_tag_file);

        #[cfg(all(target_os = "linux", feature = "io-uring"))]
        if let Some(w) = &mut self.uring_writer {
            use std::os::unix::io::AsRawFd;
            w.set_file(self.active.file.as_raw_fd());
            w.set_direct(self.effective_direct_io);
        }

        // F2: transfer ownership of the retired file into `immutable` BEFORE any
        // further fallible work (hint generation). This upholds the invariant that
        // every data file referenced by live index entries is reachable as the
        // active file or as an immutable member.
        self.immutable.push(std::sync::Arc::new(ImmutableFile {
            file: imm_file,
            file_id: old_file_id,
            #[cfg(feature = "encryption")]
            path: old_path.clone(),
            total_bytes: file_len,
            #[cfg(feature = "encryption")]
            tag_file: imm_tag_file,
        }));
        drop(old_file);

        #[cfg(feature = "encryption")]
        drop(old_tag_file);

        // Generate the hint file for the now-immutable data file. The read handle
        // / tag file are borrowed from the just-pushed immutable entry. A failure
        // here NEVER orphans the file (it is already in `immutable`): an early stop
        // (`CorruptedEntry`) is warned and the hint skipped; other errors propagate
        // but leave shard state consistent.
        if self.hints {
            #[cfg(test)]
            let force_corrupt = self.test_force_hint_corruption;
            #[cfg(not(test))]
            let force_corrupt = false;

            let imm = self
                .immutable
                .last()
                .expect("retired file was just pushed into immutable");

            let hint_result: DbResult<Vec<u8>> = if force_corrupt {
                Err(crate::error::DbError::CorruptedEntry { offset: 0 })
            } else {
                #[cfg(feature = "encryption")]
                {
                    if let (Some(cipher), Some(tag_file)) = (&self.cipher, &imm.tag_file) {
                        crate::hint::generate_hint_data_dyn_encrypted(
                            &imm.file,
                            file_len,
                            key_len,
                            cipher,
                            tag_file.as_ref(),
                            old_file_id,
                        )
                    } else {
                        crate::hint::generate_hint_data_dyn(&imm.file, file_len, key_len)
                    }
                }
                #[cfg(not(feature = "encryption"))]
                {
                    crate::hint::generate_hint_data_dyn(&imm.file, file_len, key_len)
                }
            };

            match hint_result {
                Ok(hint_data) => {
                    let hint_path = crate::hint::hint_path_for_data(&old_path);
                    crate::hint::write_hint_file(&hint_path, &hint_data)?;
                }
                Err(crate::error::DbError::CorruptedEntry { offset }) => {
                    tracing::warn!(
                        offset,
                        shard_id = _shard_id,
                        "hint generation stopped early — skipping hint file (retired file retained in immutable)"
                    );
                }
                Err(e) => return Err(e),
            }
        }

        Ok(())
    }

    pub fn add_dead_bytes(&mut self, file_id: u32, size: u64) {
        *self.dead_bytes.entry(file_id).or_insert(0) += size;
    }

    /// Allocate the next file id, returning an error if the space is exhausted.
    ///
    /// The encryption layer derives AES-GCM nonces from (file_id, page_number)
    /// in `crypto::make_page_nonce`. Wrapping past u32::MAX would reuse nonces
    /// and break GCM confidentiality and integrity guarantees.
    pub(crate) fn allocate_file_id(&mut self) -> DbResult<u32> {
        if self.next_file_id == u32::MAX {
            return Err(crate::error::DbError::Client("file_id space exhausted"));
        }
        let id = self.next_file_id;
        self.next_file_id += 1;
        Ok(id)
    }

    /// Append pre-serialized entry bytes from a replication stream.
    /// Does NOT increment the GSN counter — the entry already contains the leader's GSN.
    /// Returns `(file_id, entry_offset)` — the file the entry landed in and the byte
    /// offset within that file. Both are captured after any rotation so callers do
    /// not need to re-query `active_file_id()` separately (C2).
    #[cfg(feature = "replication")]
    pub fn append_raw_entry(
        &mut self,
        shard_id: u8,
        key_len: u16,
        data: &[u8],
    ) -> DbResult<(u32, u64)> {
        if data.len() > self.write_buf.capacity() {
            return Err(crate::error::DbError::Client(
                "replicated entry exceeds write_buffer_size",
            ));
        }

        // V-04/F4: Rotate-before-append so entry_offset fits in u32. Use the real
        // buffered position (`base_offset + len`), not the logical
        // `active.write_offset` which lags by up to a page after a force-flush
        // padding advance — the downstream apply path narrows this offset via
        // `as u32`, so an under-counted check could alias a DiskLoc.
        let next_offset = self.write_buf.base_offset + self.write_buf.len as u64;
        if next_offset + (data.len() as u64) > self.max_file_size {
            self.rotate(shard_id, key_len as usize)?;
        }

        if self.write_buf.is_full(data.len()) {
            self.flush_write_buf()?;
            if self.write_buf.is_full(data.len()) {
                // Encrypted shard retained a partial page that still blocks this
                // append after the non-force flush evicted complete pages. Force
                // a rotation to free buffer capacity (V-03).
                self.rotate(shard_id, key_len as usize)?;
            }
        }

        let file_id = self.active.file_id;
        let entry_offset = self.write_buf.append(data);
        self.active.write_offset = self.write_buf.base_offset + self.write_buf.len as u64;

        Ok((file_id, entry_offset))
    }
}

#[cfg(feature = "var-collections")]
struct BlockFileRefs {
    active_read: Option<Arc<std::fs::File>>,
    immutable_file: Option<Arc<ImmutableFile>>,
    immutable_total: u64,
    #[cfg(feature = "encryption")]
    cipher: Option<Arc<PageCipher>>,
    #[cfg(feature = "encryption")]
    tag_file: Option<Arc<TagFile>>,
}

#[cfg(feature = "var-collections")]
fn pread_decrypt_block(
    _file_id: u32,
    block_offset: u64,
    refs: &BlockFileRefs,
) -> DbResult<(AlignedBuf, bool)> {
    #[allow(unused_mut)]
    let (mut buf, _) = if let Some(file) = &refs.active_read {
        direct::pread_block(file, block_offset)?
    } else if let Some(arc) = &refs.immutable_file {
        direct::pread_block(&arc.file, block_offset)?
    } else {
        unreachable!()
    };

    let is_full_block = refs.active_read.is_none() && block_offset + 4096 <= refs.immutable_total;

    #[cfg(feature = "encryption")]
    if let Some(cipher) = &refs.cipher {
        let page_number = block_offset / 4096;
        match &refs.tag_file {
            Some(tf) => {
                let tag = tf.read_tag(page_number)?;
                cipher.decrypt_page(_file_id, page_number, &mut buf, &tag)?;
            }
            None => {
                // Active: missing tag file is a programming bug → EncryptionError.
                // Immutable: may race with compaction → StaleDiskLoc triggers retry.
                if refs.active_read.is_some() {
                    return Err(crate::error::DbError::EncryptionError(
                        "no tag file for active encrypted file_id".into(),
                    ));
                }
                return Err(crate::error::DbError::StaleDiskLoc);
            }
        }
    }

    Ok((buf, is_full_block))
}

#[cfg(feature = "var-collections")]
impl ShardInner {
    fn resolve_block_refs(&self, file_id: u32) -> DbResult<BlockFileRefs> {
        if self.active.file_id == file_id {
            Ok(BlockFileRefs {
                active_read: Some(self.active.read_file.clone()),
                immutable_file: None,
                immutable_total: 0,
                #[cfg(feature = "encryption")]
                cipher: self.cipher.clone(),
                #[cfg(feature = "encryption")]
                tag_file: self.active.tag_file.clone(),
            })
        } else {
            let arc = self
                .immutable
                .iter()
                .find(|f| f.file_id == file_id)
                .ok_or(crate::error::DbError::StaleDiskLoc)?
                .clone();
            let total = arc.total_bytes;
            #[cfg(feature = "encryption")]
            let tag_file = arc.tag_file.clone();
            Ok(BlockFileRefs {
                active_read: None,
                immutable_file: Some(arc),
                immutable_total: total,
                #[cfg(feature = "encryption")]
                cipher: self.cipher.clone(),
                #[cfg(feature = "encryption")]
                tag_file,
            })
        }
    }

    pub(crate) fn read_block_locked(
        &self,
        file_id: u32,
        block_offset: u64,
    ) -> DbResult<(AlignedBuf, bool)> {
        let refs = self.resolve_block_refs(file_id)?;
        pread_decrypt_block(file_id, block_offset, &refs)
    }

    /// Read a value from disk while the shard mutex is already held.
    ///
    /// Returns plaintext bytes when encryption is enabled. The caller is
    /// responsible for checking the write buffer and block cache first;
    /// this helper only inspects the active and immutable files.
    pub(crate) fn read_value_from_disk_locked(&self, loc: &DiskLoc) -> DbResult<Vec<u8>> {
        let len = loc.len as usize;
        let target_file_id = loc.file_id;

        if self.active.file_id == target_file_id {
            #[cfg(feature = "encryption")]
            if let Some(cipher) = &self.cipher {
                let tag_file = self.active.tag_file.as_ref().ok_or_else(|| {
                    crate::error::DbError::EncryptionError(
                        "tag file missing for active encrypted file".to_string(),
                    )
                })?;
                return crate::io::direct::pread_value_encrypted(
                    &self.active.read_file,
                    tag_file.as_ref(),
                    cipher,
                    target_file_id,
                    loc.offset as u64,
                    len,
                );
            }
            return crate::io::direct::pread_value(&self.active.read_file, loc.offset as u64, len);
        }

        let imm = self
            .immutable
            .iter()
            .find(|f| f.file_id == target_file_id)
            .ok_or(crate::error::DbError::StaleDiskLoc)?;

        #[cfg(feature = "encryption")]
        if let Some(cipher) = &self.cipher {
            let tag_file = imm
                .tag_file
                .as_ref()
                .ok_or(crate::error::DbError::StaleDiskLoc)?;
            return crate::io::direct::pread_value_encrypted(
                &imm.file,
                tag_file.as_ref(),
                cipher,
                target_file_id,
                loc.offset as u64,
                len,
            );
        }
        crate::io::direct::pread_value(&imm.file, loc.offset as u64, len)
    }

    /// True when `loc`'s byte range straddles the write buffer's `base_offset`:
    /// the prefix `[loc.offset, base_offset)` has already been flushed to disk
    /// (page-granular non-force flush in `page_aligned` mode) while the tail
    /// `[base_offset, loc.offset + loc.len)` is still buffered in RAM. Assumes
    /// the caller has verified `loc` targets the **active** file — only then are
    /// the sub-`base_offset` bytes guaranteed to be on disk.
    pub(crate) fn straddles_write_buf(&self, loc: &DiskLoc) -> bool {
        let base = self.write_buf.base_offset;
        let start = loc.offset as u64;
        let end = start + loc.len as u64;
        start < base && end > base
    }

    /// Read a value whose range straddles the write buffer's `base_offset`
    /// (see [`ShardInner::straddles_write_buf`]). The prefix is read from disk
    /// via the encryption-aware locked helper — those pages were flushed by a
    /// page-granular non-force flush, so their tags exist — and the tail is
    /// copied from the write buffer, which holds `[base_offset, ..)` starting at
    /// `buf[0]`. Both halves are stable under the shard lock. Returns plaintext.
    pub(crate) fn read_straddling_value_locked(&self, loc: &DiskLoc) -> DbResult<Vec<u8>> {
        debug_assert_eq!(self.active.file_id, loc.file_id);
        debug_assert!(self.straddles_write_buf(loc));

        let base = self.write_buf.base_offset;
        let prefix_len = (base - loc.offset as u64) as usize;
        let tail_len = loc.len as usize - prefix_len;

        // Prefix: already-flushed bytes on disk (encryption-aware). The prefix
        // ends exactly at `base_offset` (page-aligned), so it spans only sealed,
        // tagged pages.
        let prefix_loc = DiskLoc::new(loc.file_id, loc.offset, prefix_len as u32);
        let mut out = self.read_value_from_disk_locked(&prefix_loc)?;

        // Tail: still buffered, starting exactly at `base_offset`.
        let tail = self
            .write_buf
            .read(base, tail_len)
            .ok_or(crate::error::DbError::StaleDiskLoc)?;
        out.extend_from_slice(tail);
        Ok(out)
    }
}

#[cfg(test)]
#[allow(unused)]
impl Shard {
    /// Force `next_file_id` to a target value. Used by tests that need
    /// to exercise the file-id space past u16 without writing 65536 files.
    pub(crate) fn set_next_file_id(&self, id: u32) {
        sync::lock(&self.inner).next_file_id = id;
    }

    /// Trigger an immediate shard rotation (flush + new file). Used by tests.
    ///
    /// `key_len` is the byte width of the keys written through this shard;
    /// it is forwarded to the hint generator so the post-rotation hint file
    /// can be parsed correctly by recovery. Callers must pass the same value
    /// the surrounding collection uses for its keys (e.g. `1` for `b"k"`,
    /// `8` for `[u8; 8]`).
    pub(crate) fn rotate_active_for_test(&self, key_len: usize) -> DbResult<()> {
        sync::lock(&self.inner).rotate(self.id, key_len)
    }
}

#[cfg(test)]
#[allow(unused_imports)]
mod tests {
    use super::*;
    use crate::entry::{EntryHeader, entry_size};
    use crate::error::DbError;
    use std::mem::size_of;
    use std::sync::atomic::AtomicU64;
    use tempfile::tempdir;
    use zerocopy::FromBytes;

    #[cfg(feature = "var-collections")]
    fn open_test_shard(dir: &std::path::Path) -> Shard {
        let gsn = Arc::new(AtomicU64::new(0));
        Shard::open(
            0,
            dir,
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            gsn,
        )
        .expect("open test shard")
    }

    #[cfg(feature = "replication")]
    fn append_test_entry(shard: &Shard, key: u64, value: &[u8]) {
        crate::sync::lock(&shard.inner)
            .append_entry(shard.id, &key.to_be_bytes(), value, false)
            .unwrap();
    }

    #[cfg(feature = "replication")]
    fn shard_with_rotated_entries() -> (tempfile::TempDir, Shard) {
        let dir = tempfile::tempdir().unwrap();
        let shard = open_test_shard(dir.path());
        shard.gsn.store(1, Ordering::Relaxed);
        append_test_entry(&shard, 1, b"one");
        shard.rotate_active_for_test(8).unwrap();
        (dir, shard)
    }

    #[cfg(feature = "replication")]
    fn shard_with_entries_in_active_file() -> (tempfile::TempDir, Shard) {
        let dir = tempfile::tempdir().unwrap();
        let shard = open_test_shard(dir.path());
        shard.gsn.store(1, Ordering::Relaxed);
        append_test_entry(&shard, 1, b"one");
        (dir, shard)
    }

    #[cfg(feature = "replication")]
    fn rotate_and_append(shard: &Shard) {
        shard.rotate_active_for_test(8).unwrap();
        append_test_entry(shard, 2, b"two");
    }

    #[cfg(all(feature = "encryption", feature = "replication"))]
    fn open_encrypted_test_shard(dir: &std::path::Path) -> Shard {
        Shard::open_encrypted(
            0,
            dir,
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            Some(Arc::new(PageCipher::new(&[0x42; 32]).unwrap())),
            Arc::new(AtomicU64::new(1)),
        )
        .unwrap()
    }

    #[cfg(feature = "replication")]
    #[test]
    fn snapshot_reads_source_after_path_unlink() {
        let (dir, shard) = shard_with_rotated_entries();
        let snapshot = shard.catchup_snapshot().unwrap();
        let source = snapshot.files.first().unwrap().clone();
        std::fs::remove_file(dir.path().join("000001.data")).unwrap();
        let header = source.read_exact_at(0, size_of::<EntryHeader>()).unwrap();
        assert_eq!(EntryHeader::read_from_bytes(&header).unwrap().sequence(), 1);
    }

    #[cfg(feature = "replication")]
    #[test]
    fn snapshot_active_boundary_does_not_move_after_rotation() {
        let (_dir, shard) = shard_with_entries_in_active_file();
        let snapshot = shard.catchup_snapshot().unwrap();
        let captured_end = snapshot.files.last().unwrap().readable_end;
        rotate_and_append(&shard);
        assert_eq!(snapshot.files.last().unwrap().readable_end, captured_end);
        assert!(
            snapshot
                .files
                .last()
                .unwrap()
                .read_exact_at(captured_end, 1)
                .is_err()
        );
    }

    #[cfg(all(feature = "encryption", feature = "replication"))]
    #[test]
    fn encrypted_snapshot_retains_data_and_tag_handles_after_unlink() {
        let dir = tempfile::tempdir().unwrap();
        let shard = open_encrypted_test_shard(dir.path());
        append_test_entry(&shard, 1, b"encrypted");
        shard.rotate_active_for_test(8).unwrap();
        let snapshot = shard.catchup_snapshot().unwrap();
        let source = snapshot.files.first().unwrap();
        std::fs::remove_file(dir.path().join("000001.data")).unwrap();
        std::fs::remove_file(dir.path().join("000001.tags")).unwrap();
        let bytes = source.read_exact_at(0, entry_size(8, 9) as usize).unwrap();
        assert_eq!(
            EntryHeader::read_from_bytes(&bytes[..16])
                .unwrap()
                .sequence(),
            1
        );
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn read_block_returns_stale_for_unknown_file_id() {
        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());

        // The freshly opened shard has only file_id == 1 as active. file_id 9999
        // exists in neither active nor immutable.
        match shard.read_block(9999, 0) {
            Err(DbError::StaleDiskLoc) => {}
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
            Ok(_) => panic!("expected StaleDiskLoc, got Ok"),
        }
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn read_value_returns_stale_for_unknown_file_id() {
        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());
        match shard.read_value(9999, 0, 128) {
            Err(DbError::StaleDiskLoc) => {}
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
            Ok(_) => panic!("expected StaleDiskLoc, got Ok"),
        }
    }

    #[cfg(all(feature = "encryption", feature = "var-collections"))]
    #[test]
    fn read_block_returns_stale_when_immutable_missing_under_encryption() {
        use crate::crypto::PageCipher;

        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Some(Arc::new(
            PageCipher::new(&[0x42; 32]).expect("create cipher"),
        ));
        let shard = Shard::open_encrypted(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            cipher,
            gsn,
        )
        .expect("open encrypted shard");

        // The shard has only active file_id=1 right now. read_block on a
        // non-existent file_id must hit Change A and return Stale before any
        // encrypted-decode path runs.
        match shard.read_block(42, 0) {
            Err(DbError::StaleDiskLoc) => {}
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
            Ok(_) => panic!("expected StaleDiskLoc, got Ok"),
        }
    }

    #[cfg(feature = "var-collections")]
    #[test]
    fn read_value_from_disk_locked_returns_stale_for_unknown_file_id() {
        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());

        let fake = DiskLoc::new(9999, 0, 0);
        let inner = shard.lock();
        match inner.read_value_from_disk_locked(&fake) {
            Err(DbError::StaleDiskLoc) => {}
            Ok(_) => panic!("expected StaleDiskLoc, got Ok"),
            Err(e) => panic!("expected StaleDiskLoc, got Err({e})"),
        }
    }

    /// Pin `read_block` (immutable-file path) for file_id values above u16::MAX.
    /// Before the file_id u32 widening these would either panic or silently read
    /// aliased data because the high bits were truncated on storage.
    #[cfg(feature = "var-collections")]
    #[test]
    fn read_block_with_file_id_above_u16() {
        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());

        // Jump next_file_id well past u16::MAX and force a rotation so that
        // the current active file gets a >u16 id.
        shard.set_next_file_id(70_000);
        shard.rotate_active_for_test(3).expect("first rotate");
        assert!(
            shard.active_file_id() >= 70_000,
            "active_file_id should be >= 70_000 after rotation"
        );

        let key = b"abc";
        let value = vec![0xABu8; 256];
        let (disk, _gsn) = shard
            .lock()
            .append_entry(0, key, &value, false)
            .expect("append entry");
        assert!(
            disk.file_id > u16::MAX as u32,
            "DiskLoc.file_id must be above u16::MAX"
        );

        // Flush + rotate so the written file becomes immutable (read_block path).
        shard.flush().expect("flush");
        shard.rotate_active_for_test(3).expect("second rotate");

        // read_block reads from the now-immutable file at a 4096-byte aligned offset.
        let block_offset = disk.offset as u64 & !4095;
        let (block, _) = shard
            .read_block(disk.file_id, block_offset)
            .expect("read_block");
        let start = (disk.offset & 4095) as usize;
        let end = start + value.len();
        assert_eq!(&block[start..end], &value[..]);
    }

    /// Pin `write_buf.read` (active write-buffer path) for file_id values above u16::MAX.
    #[cfg(feature = "var-collections")]
    #[test]
    fn write_buffer_read_with_file_id_above_u16() {
        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());

        // Bump file_id past u16 range and rotate so active file gets the new id.
        shard.set_next_file_id(70_000);
        shard.rotate_active_for_test(1).expect("rotate");
        assert!(shard.active_file_id() >= 70_000);

        let value = vec![0x5Au8; 200];
        let (disk, _gsn) = shard
            .lock()
            .append_entry(0, b"k", &value, false)
            .expect("append");
        assert!(
            disk.file_id > u16::MAX as u32,
            "DiskLoc.file_id must be above u16::MAX"
        );

        // Data is still in the write buffer (not yet flushed). Verify that
        // write_buf.read finds it at the correct absolute file offset.
        let inner = shard.lock();
        let bytes = inner
            .write_buf
            .read(disk.offset as u64, disk.len as usize)
            .expect("write-buf read must succeed for unflushed entry");
        assert_eq!(bytes, &value[..]);
    }

    #[test]
    fn rotate_errors_when_file_id_exhausted() {
        let tmp = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(1));
        let shard = Shard::open(
            0,
            tmp.path(),
            16 * 4096,
            8192,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            gsn,
        )
        .unwrap();
        shard.set_next_file_id(u32::MAX);
        let res = shard.rotate_active_for_test(4);
        match res {
            Err(DbError::Client(msg)) => {
                assert!(
                    msg.contains("file_id"),
                    "expected file_id error, got: {msg}"
                );
            }
            other => panic!("expected DbError::Client, got {other:?}"),
        }
    }

    #[test]
    fn open_errors_when_active_id_is_max() {
        let tmp = tempdir().unwrap();
        let path = tmp.path().join(format!("{}.data", u32::MAX));
        std::fs::write(&path, b"").unwrap();
        let gsn = Arc::new(AtomicU64::new(1));
        let res = Shard::open(
            0,
            tmp.path(),
            16 * 4096,
            8192,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            gsn,
        );
        match res {
            Err(DbError::Client(msg)) => {
                assert!(
                    msg.contains("file_id"),
                    "expected file_id error, got: {msg}"
                );
            }
            Ok(_) => panic!("expected DbError::Client, got Ok"),
            Err(e) => panic!("expected DbError::Client, got Err({e})"),
        }
    }

    /// `flush_for_replication_catchup` in encrypted mode must pad the trailing
    /// partial page to a 4096-byte boundary so that the encrypted
    /// `ShardLogReader` can decrypt every entry up to `shard.gsn()`.
    ///
    /// The test asserts the *contrast* between the two flush modes:
    /// 1. `flush_buf` (NonForce) must leave the small entry in the write buffer
    ///    (zero bytes on disk for an encrypted shard with one sub-4096 entry).
    /// 2. `flush_for_replication_catchup` (ForceAdvance) must pad + encrypt the
    ///    trailing page so the file grows to a 4096-aligned size.
    ///
    /// Without this contrast the assertion `file_len % 4096 == 0` would also
    /// pass if the method were wired to NonForce (since 0 is page-aligned).
    #[cfg(all(feature = "encryption", feature = "var-collections"))]
    #[test]
    fn flush_for_replication_catchup_pads_encrypted_trailing_page() {
        use crate::crypto::PageCipher;

        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Some(Arc::new(
            PageCipher::new(&[0xAB; 32]).expect("create cipher"),
        ));
        let shard = Shard::open_encrypted(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            cipher,
            gsn,
        )
        .expect("open encrypted shard");

        // Append a small entry (much smaller than 4096 bytes).
        // EntryHeader (16) + key (3) + value (5) padded to 8 = 32 bytes.
        {
            let mut inner = sync::lock(&shard.inner);
            inner
                .append_entry(0, b"key", b"value", false)
                .expect("append_entry");
        }

        // The active file is <shard_dir>/000001.data.
        let data_path = dir.path().join("000001.data");

        // Step 1 (NonForce): partial encrypted page stays in the buffer.
        // The encrypted flush path only writes complete 4096-byte pages, so
        // a single sub-page entry must NOT reach disk under NonForce.
        shard.flush_buf().expect("flush_buf");
        let len_after_nonforce = std::fs::metadata(&data_path).expect("metadata").len();
        assert_eq!(
            len_after_nonforce, 0,
            "NonForce encrypted flush must leave partial page in buffer (no whole page yet), got {len_after_nonforce}"
        );

        // Step 2 (ForceAdvance via flush_for_replication_catchup):
        // the trailing partial page must be padded with zeros, encrypted,
        // and written to disk — so the file becomes 4096-aligned and non-empty.
        shard
            .flush_for_replication_catchup()
            .expect("flush_for_replication_catchup");
        let len_after_force = std::fs::metadata(&data_path).expect("metadata").len();
        assert!(
            len_after_force > 0,
            "ForceAdvance must flush the padded page to disk, got 0 bytes"
        );
        assert_eq!(
            len_after_force % 4096,
            0,
            "encrypted file length must be 4096-aligned after ForceAdvance, got {len_after_force}"
        );
        // Single sub-4096 entry → exactly one padded page on disk.
        assert_eq!(
            len_after_force, 4096,
            "expected exactly one 4096-byte padded page, got {len_after_force}"
        );
    }

    #[cfg(feature = "encryption")]
    #[test]
    fn apply_recovery_tail_encrypted_advances_to_high_watermark() {
        use crate::crypto::PageCipher;

        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Some(Arc::new(PageCipher::new(&[0xCD; 32]).unwrap()));
        let shard = Shard::open_encrypted(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            cipher,
            gsn,
        )
        .expect("open encrypted shard");

        // Две суб-страничные записи, каждая force-flush'ится → две запечатанные
        // 4096-байтные страницы в 000001.data и два тега в 000001.tags.
        {
            let mut inner = sync::lock(&shard.inner);
            inner
                .append_entry(0, b"k1", b"v1", false)
                .expect("append k1");
        }
        shard.flush().expect("flush 1");
        {
            let mut inner = sync::lock(&shard.inner);
            inner
                .append_entry(0, b"k2", b"v2", false)
                .expect("append k2");
        }
        shard.flush().expect("flush 2");

        let data_len = std::fs::metadata(dir.path().join("000001.data"))
            .unwrap()
            .len();
        assert_eq!(data_len, 8192, "expected two sealed 4096-byte pages");

        // Имитируем recovery-скан, провалидировавший только первую страницу.
        let tail = crate::recovery::ActiveTail {
            shard_idx: 0,
            file_id: 1,
            last_valid_offset: 4096,
        };
        shard
            .apply_recovery_tail(&tail)
            .expect("apply_recovery_tail");

        // Без F6 база осталась бы 4096, и следующий append перезапечатал бы
        // страницу 1 под тем же (subkey, nonce). Должна прыгнуть на 8192.
        let base = sync::lock(&shard.inner).active.write_offset;
        assert_eq!(
            base, 8192,
            "F6: post-recovery base must clear all sealed pages"
        );
    }

    #[test]
    fn reopen_page_aligned_appends_at_page_boundary() {
        use crate::recovery::recover_const_tree;
        use crate::skiplist::SkipList;
        use crate::skiplist::node::ConstNode;

        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        const WB: usize = 8192;

        // Phase 1: write one entry, force-flush (pads to page), close.
        {
            let shard = Shard::open(
                0,
                dir.path(),
                1 << 20,
                WB,
                false,
                true,
                false,
                crate::config::IoBackend::default(),
                gsn.clone(),
            )
            .expect("open page-aligned shard");
            {
                let mut inner = sync::lock(&shard.inner);
                inner
                    .append_entry(0, b"key", b"value", false)
                    .expect("append");
            }
            shard.flush().expect("final flush");
        }

        // Phase 2: reopen, run recovery, apply tail, assert alignment + append works.
        {
            let shard = Shard::open(
                0,
                dir.path(),
                1 << 20,
                WB,
                false,
                true,
                false,
                crate::config::IoBackend::default(),
                gsn.clone(),
            )
            .expect("reopen page-aligned shard");

            let index = SkipList::<ConstNode<[u8; 3], 5>>::new(false);
            let outcome = recover_const_tree::<[u8; 3], 5>(
                &[dir.path()],
                &[0],
                &index,
                false,
                #[cfg(feature = "encryption")]
                None,
            )
            .expect("recover");

            assert_eq!(outcome.active_tails.len(), 1);
            let tail = &outcome.active_tails[0];
            assert!(
                tail.last_valid_offset < 4096,
                "expected sub-page last_valid_offset, got {}",
                tail.last_valid_offset
            );

            shard
                .apply_recovery_tail(tail)
                .expect("apply_recovery_tail");

            {
                let inner = sync::lock(&shard.inner);
                assert!(inner.page_aligned);
                assert_eq!(
                    inner.write_buf.base_offset % 4096,
                    0,
                    "base_offset must be page-aligned after apply_recovery_tail"
                );
            }

            {
                let mut inner = sync::lock(&shard.inner);
                inner
                    .append_entry(0, b"k2!", b"val2!", false)
                    .expect("append after recovery");
            }
            shard.flush_buf().expect("flush_buf");
            shard.flush().expect("final flush");
        }
    }

    #[test]
    fn plain_flush_is_page_granular_when_page_aligned() {
        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let shard = Shard::open(
            0,
            dir.path(),
            1 << 20,
            8192,
            false,
            true,
            false,
            crate::config::IoBackend::default(),
            gsn,
        )
        .expect("open page-aligned shard");

        {
            let mut inner = sync::lock(&shard.inner);
            inner
                .append_entry(0, b"key", b"value", false)
                .expect("append_entry");
        }

        let data_path = dir.path().join("000001.data");

        shard.flush_buf().expect("flush_buf");
        let len_after_nonforce = std::fs::metadata(&data_path).expect("metadata").len();
        assert_eq!(
            len_after_nonforce, 0,
            "page-aligned NonForce flush must leave partial page in buffer, got {len_after_nonforce}"
        );

        shard.flush().expect("flush");
        let len_after_force = std::fs::metadata(&data_path).expect("metadata").len();
        assert_eq!(
            len_after_force, 4096,
            "page-aligned ForceAdvance must pad trailing page to 4096 bytes, got {len_after_force}"
        );
    }

    /// Encrypted flush with `direct=true` must use an `AlignedBuf` so O_DIRECT
    /// writes succeed on supporting filesystems. Exercises multi-page encrypted
    /// flush, read-back, and reopen recovery.
    #[cfg(feature = "encryption")]
    #[test]
    fn encrypted_direct_roundtrip() {
        use crate::crypto::PageCipher;

        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let cipher = Some(Arc::new(
            PageCipher::new(&[0xCD; 32]).expect("create cipher"),
        ));

        // Large enough to span multiple 4096-byte encrypted pages when flushed.
        let value: Vec<u8> = (0..9000).map(|i| (i % 251) as u8).collect();
        let expected = value.clone();

        let disk = {
            let shard = Shard::open_encrypted(
                0,
                dir.path(),
                1 << 20,
                1 << 20,
                false,
                true,
                true,
                crate::config::IoBackend::default(),
                cipher.clone(),
                gsn.clone(),
            )
            .expect("open encrypted direct shard");
            assert!(shard.effective_direct_io());

            let (disk, _gsn) = sync::lock(&shard.inner)
                .append_entry(0, b"key", &value, false)
                .expect("append_entry");
            shard.flush().expect("flush");
            disk
        };

        {
            let shard = Shard::open_encrypted(
                0,
                dir.path(),
                1 << 20,
                1 << 20,
                false,
                true,
                false,
                crate::config::IoBackend::default(),
                cipher,
                gsn,
            )
            .expect("reopen encrypted direct shard");

            let inner = sync::lock(&shard.inner);
            let got = inner
                .read_value_from_disk_locked(&disk)
                .expect("read encrypted value after reopen");
            assert_eq!(got, expected);
        }
    }

    /// `flush_for_replication_catchup` in plain mode writes exactly the entry
    /// bytes to disk (no padding). Verifies the wrapper does not accidentally
    /// inject padding in the unencrypted path.
    #[cfg(feature = "var-collections")]
    #[test]
    fn flush_for_replication_catchup_plain_mode_flushes() {
        let dir = tempdir().unwrap();
        let shard = open_test_shard(dir.path());

        // EntryHeader (16) + key (1) + value (1) = 18, padded to 8-byte boundary = 24.
        let expected_entry_size = crate::entry::entry_size(b"k".len(), b"v".len() as u32);
        assert_eq!(expected_entry_size, 24);

        {
            let mut inner = sync::lock(&shard.inner);
            inner
                .append_entry(0, b"k", b"v", false)
                .expect("append_entry");
        }

        shard
            .flush_for_replication_catchup()
            .expect("flush_for_replication_catchup plain");

        let data_path = dir.path().join("000001.data");
        let file_len = std::fs::metadata(&data_path).expect("metadata").len();

        // Plain mode writes the entry bytes verbatim — no page padding.
        assert_eq!(
            file_len, expected_entry_size,
            "plain mode must write exactly entry bytes (no padding); expected {expected_entry_size}, got {file_len}"
        );
    }

    /// F2: when hint generation stops early (e.g. `CorruptedEntry` while scanning
    /// the just-retired file), `rotate()` must still transfer that file into
    /// `immutable`. Before the fix an early `return Ok(())` left the file orphaned
    /// — neither active nor immutable — so every live `DiskLoc` into it read back
    /// as `StaleDiskLoc` until process restart.
    #[test]
    fn rotate_retains_retired_file_when_hint_gen_stops_early() {
        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        let shard = Shard::open(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            true, // hints enabled — exercises the hint-generation path in rotate()
            false,
            false,
            crate::config::IoBackend::default(),
            gsn,
        )
        .expect("open shard with hints");

        let value = vec![0xABu8; 128];
        let disk = {
            let mut inner = sync::lock(&shard.inner);
            let (disk, _gsn) = inner.append_entry(0, b"k", &value, false).expect("append");
            disk
        };
        let old_file_id = disk.file_id; // == 1

        // Force the next rotation's hint generation to hit the early-stop path.
        sync::lock(&shard.inner).test_force_hint_corruption = true;

        shard.rotate_active_for_test(1).expect("rotate");

        // Invariant: the retired file is reachable as an immutable member.
        let ids = shard.file_ids();
        assert!(
            ids.contains(&old_file_id),
            "retired file {old_file_id} must be retained in immutable, got file_ids {ids:?}"
        );

        // And a live DiskLoc into it must still read (not StaleDiskLoc). The
        // rotation flushed the buffer, so the value is on disk in the now-immutable
        // file.
        #[cfg(feature = "var-collections")]
        {
            let inner = shard.lock();
            let got = inner
                .read_value_from_disk_locked(&disk)
                .expect("read retired-file value must succeed, not StaleDiskLoc");
            assert_eq!(got, value);
        }
    }

    /// F4: the rotate-before-append check must use the real buffered position
    /// (`base_offset + len`), not the logical `active.write_offset`. After a
    /// force-flush pads a partial page, `base_offset` jumps to the next page while
    /// `write_offset` lags at the logical end — using the stale value under-counts
    /// the append position and can push `value_offset` past `u32::MAX`.
    #[test]
    fn rotate_check_accounts_for_force_flush_padding_gap() {
        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));
        const WB: usize = 8192;
        const MAX: u64 = 8192; // two pages
        // Page-aligned mode (direct_io=true) so a force-flush pads to a page.
        let shard = Shard::open(
            0,
            dir.path(),
            MAX,
            WB,
            false,
            true, // direct_io → page_aligned
            false,
            crate::config::IoBackend::default(),
            gsn,
        )
        .expect("open page-aligned shard");

        // Entry A: small, sub-page.
        {
            let mut inner = sync::lock(&shard.inner);
            inner
                .append_entry(0, b"k", &[0u8; 100], false)
                .expect("append A");
        }
        // Force-flush: pads the partial page, advancing base_offset to 4096 while
        // active.write_offset stays at the ~130-byte logical end.
        shard.flush().expect("force flush");
        {
            let inner = sync::lock(&shard.inner);
            assert_eq!(
                inner.write_buf.base_offset, 4096,
                "force flush must advance base_offset to the next page"
            );
            assert!(
                inner.active.write_offset < 4096,
                "logical write_offset must lag behind the padded base_offset (got {})",
                inner.active.write_offset
            );
        }

        // Entry B: value ~5000 bytes. needed ≈ 5024.
        //   stale check:  write_offset(~130) + 5024 = ~5154  <= 8192 → NO rotate (bug)
        //   correct check: base_offset(4096) + 5024 = 9120    >  8192 → rotate
        let value_b = vec![0x11u8; 5000];
        let loc_b = {
            let mut inner = sync::lock(&shard.inner);
            let (loc, _gsn) = inner
                .append_entry(0, b"k", &value_b, false)
                .expect("append B");
            loc
        };

        assert_eq!(
            loc_b.file_id, 2,
            "append past max_file_size (accounting for padding gap) must rotate to a new file, \
             got file_id {} (offset {})",
            loc_b.file_id, loc_b.offset
        );
        assert!(
            loc_b.offset < 4096,
            "after rotation the value must land near the start of the new file, got {}",
            loc_b.offset
        );
    }

    /// F5: creating a new data file during `rotate()` and on first open must
    /// fsync the shard directory so the new file's directory entry is durable.
    /// Verified via the `sync_dir` call counter (a real power-loss is not
    /// reproducible in-process) plus a rotate→reopen round-trip.
    #[test]
    fn rotate_and_first_open_fsync_shard_dir() {
        use crate::io::direct::SYNC_DIR_CALLS;
        use std::sync::atomic::Ordering;

        let dir = tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(0));

        let before_open = SYNC_DIR_CALLS.load(Ordering::Relaxed);
        let shard = Shard::open(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            gsn.clone(),
        )
        .expect("open shard (create branch)");
        let after_open = SYNC_DIR_CALLS.load(Ordering::Relaxed);
        assert!(
            after_open > before_open,
            "first open (create branch) must fsync the shard dir"
        );

        shard.rotate_active_for_test(1).expect("rotate");
        let after_rotate = SYNC_DIR_CALLS.load(Ordering::Relaxed);
        assert!(
            after_rotate > after_open,
            "rotate() must fsync the shard dir after creating the new data file"
        );

        // Sanity: the newly created active file exists and a reopen sees every file.
        let ids = shard.file_ids();
        assert_eq!(
            ids,
            vec![1, 2],
            "expected file 1 (immutable) + file 2 (active)"
        );
        drop(shard);

        let reopened = Shard::open(
            0,
            dir.path(),
            1 << 20,
            64 * 1024,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            gsn,
        )
        .expect("reopen shard");
        let reopened_ids = reopened.file_ids();
        assert!(
            reopened_ids.contains(&2),
            "reopen must see the rotated-in file 2, got {reopened_ids:?}"
        );
    }
}

#[cfg(test)]
mod append_offset_tests {
    use super::*;
    use std::sync::Arc;
    use std::sync::atomic::AtomicU64;

    #[test]
    fn rotate_before_append_at_u32_max_boundary() {
        let tmp = tempfile::tempdir().unwrap();
        let gsn = Arc::new(AtomicU64::new(1));
        // max_file_size = u32::MAX (valid per Config::validate), write_buf = 8192.
        let shard = Shard::open(
            0,
            tmp.path(),
            u32::MAX as u64,
            8192,
            false,
            false,
            false,
            crate::config::IoBackend::default(),
            gsn,
        )
        .unwrap();
        // Bring active.write_offset close to the boundary: 100 bytes before u32::MAX.
        {
            let mut inner = sync::lock(&shard.inner);
            inner.active.write_offset = u32::MAX as u64 - 100;
            inner.write_buf.reset(u32::MAX as u64 - 100);
        }
        let key = b"k";
        let value = vec![0u8; 200];
        let (loc, _gsn) = {
            let mut inner = sync::lock(&shard.inner);
            inner.append_entry(0, key, &value, false).unwrap()
        };
        assert_eq!(loc.file_id, 2, "expected rotation to new file");
        assert!(loc.offset < 4096, "expected offset near start of new file");
    }
}