luci/storage/

single_file.rs

use std::fs::{File, OpenOptions};
#[cfg(unix)]
use std::os::unix::fs::FileExt;
use std::path::Path;
use std::sync::Arc;

use crate::core::{FieldId, LuciError, Result, SegmentId};

use crate::storage::allocator::BlockAllocator;
use crate::storage::block::{BLOCK_SIZE, Extent, HEADER_SIZE};
use crate::storage::directory::{MetadataSnapshot, SegmentEntry, VectorIndexEntry};
use crate::storage::header::{FileHeader, RootPointer, xxh3_checksum};

/// Production storage backend: all data in a single `.luci` file using the
/// block allocator and two-root-pointer atomic commit.
///
/// See [[architecture-storage-format]] for the full design and [[architecture-storage-format#Atomic Commit Protocol]]
/// for the commit sequence.
///
/// # Crash Safety
///
/// At any point, both root pointers reference intact (non-overwritten) metadata
/// blocks. A crash mid-commit is recovered by falling back to the still-valid
/// root. Uncommitted segment writes are orphaned blocks that the allocator does
/// not reference — they are harmlessly reclaimed on the next file open.
#[cfg(unix)]
pub struct SingleFileDirectory {
    file: Arc<File>,
    lock: crate::storage::lock::FileLock,
    header: FileHeader,
    allocator: BlockAllocator,
    /// Last committed metadata — the source of truth for reads.
    committed: MetadataSnapshot,
    /// Segments written but not yet committed.
    pending_segments: Vec<SegmentEntry>,
    /// Segments to remove on next commit.
    pending_removals: Vec<SegmentId>,
    /// Vector indexes written but not yet committed. Replaces any
    /// already-committed entry for the same `FieldId` on commit. See
    /// [[global-vector-indices]].
    pending_vector_indexes: Vec<VectorIndexEntry>,
    /// Vector indexes to remove on next commit.
    pending_vector_index_removals: Vec<FieldId>,
    /// Monotonically increasing commit counter.
    generation: u64,
    /// Whether user metadata has been modified since last commit.
    metadata_dirty: bool,
    /// Timeout for acquiring the cross-process write lock. Default: 5 seconds.
    write_timeout: std::time::Duration,
}

#[cfg(unix)]
impl SingleFileDirectory {
    /// Create a new `.luci` file at the given path.
    ///
    /// Fails if a file already exists at `path`.
    pub fn create(path: impl AsRef<Path>) -> Result<Self> {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .create_new(true)
            .open(path.as_ref())?;

        // Acquire SHARED lock for cross-process coordination.
        let mut lock = crate::storage::lock::FileLock::new(&file);
        lock.lock_shared()?;

        let file = Arc::new(file);
        let header = FileHeader::new();
        file.write_all_at(&header.to_bytes(), 0)?;
        file.sync_all()?;

        Ok(Self {
            file,
            lock,
            header,
            allocator: BlockAllocator::new(),
            committed: MetadataSnapshot::empty(),
            pending_segments: Vec::new(),
            pending_removals: Vec::new(),
            pending_vector_indexes: Vec::new(),
            pending_vector_index_removals: Vec::new(),
            generation: 0,
            metadata_dirty: false,
            write_timeout: std::time::Duration::from_secs(5),
        })
    }

    /// Open an existing `.luci` file, performing crash recovery.
    ///
    /// Validates root pointers, loads metadata from the best available root,
    /// and reconstructs the block allocator. If the active root's checksum is
    /// invalid (torn write), falls back to the inactive root and repairs the
    /// header.
    ///
    /// See [[architecture-storage-format#Crash Recovery]].
    pub fn open(path: impl AsRef<Path>) -> Result<Self> {
        let file = OpenOptions::new()
            .read(true)
            .write(true)
            .open(path.as_ref())?;

        // Acquire SHARED lock for cross-process coordination.
        let mut lock = crate::storage::lock::FileLock::new(&file);
        lock.lock_shared()?;

        let file = Arc::new(file);

        // Read header.
        let mut header_buf = [0u8; HEADER_SIZE as usize];
        file.read_exact_at(&mut header_buf, 0)?;
        let mut header = FileHeader::from_bytes(&header_buf)?;

        // Load metadata with fallback.
        let (committed, used_inactive) = load_metadata(&file, &header)?;

        // If we fell back to the inactive root, repair the header so the
        // valid root is now marked active.
        if used_inactive {
            header.active_root = header.active_root.inactive();
            file.write_all_at(&header.to_bytes(), 0)?;
            file.sync_all()?;
        }

        // Reconstruct allocator from committed state.
        let allocator =
            BlockAllocator::from_state(committed.total_blocks, committed.free_list.clone());

        let generation = committed
            .segments
            .iter()
            .map(|s| s.generation)
            .max()
            .unwrap_or(0);

        Ok(Self {
            file,
            lock,
            header,
            allocator,
            committed,
            pending_segments: Vec::new(),
            pending_removals: Vec::new(),
            pending_vector_indexes: Vec::new(),
            pending_vector_index_removals: Vec::new(),
            generation,
            metadata_dirty: false,
            write_timeout: std::time::Duration::from_secs(5),
        })
    }

    /// Return a shared reference to the underlying file handle.
    ///
    /// Used by the reader path to avoid opening a separate fd,
    /// which would break `fcntl` lock semantics (closing any fd
    /// releases all locks for the process on that inode).
    ///
    /// See [[architecture-cross-process-locking#Critical Constraint]].
    pub fn file_handle(&self) -> Arc<File> {
        self.file.clone()
    }

    /// Set the timeout for acquiring the cross-process write lock.
    ///
    /// Default: 5 seconds. If another process holds the write lock,
    /// retries with exponential backoff until the timeout expires,
    /// then returns `WriterLocked`.
    pub fn set_write_timeout(&mut self, timeout: std::time::Duration) {
        self.write_timeout = timeout;
    }

    /// Open a read-only view using an existing file handle.
    ///
    /// Does not open a new fd or acquire any locks. The caller's
    /// process already holds the appropriate lock via the writer's fd.
    /// Used by `refresh_reader()` to reload segment data without
    /// breaking `fcntl` lock semantics.
    pub fn open_from_handle(file: Arc<File>) -> Result<Self> {
        let mut header_buf = [0u8; HEADER_SIZE as usize];
        file.read_exact_at(&mut header_buf, 0)?;
        let header = FileHeader::from_bytes(&header_buf)?;

        let (committed, _used_inactive) = load_metadata(&file, &header)?;
        // Don't repair header — this is a read-only snapshot.

        let allocator =
            BlockAllocator::from_state(committed.total_blocks, committed.free_list.clone());

        let generation = committed
            .segments
            .iter()
            .map(|s| s.generation)
            .max()
            .unwrap_or(0);

        // Create a no-op lock (UNLOCKED) — locking is managed by the
        // writer's fd, not this reader view.
        let lock = crate::storage::lock::FileLock::new(&file);

        Ok(Self {
            file,
            lock,
            header,
            allocator,
            committed,
            pending_segments: Vec::new(),
            pending_removals: Vec::new(),
            pending_vector_indexes: Vec::new(),
            pending_vector_index_removals: Vec::new(),
            generation,
            metadata_dirty: false,
            write_timeout: std::time::Duration::from_secs(5),
        })
    }

    /// Acquire RESERVED lock and refresh allocator from disk.
    ///
    /// Called automatically before the first write in a session. Ensures
    /// the block allocator reflects the latest committed state on disk
    /// (another process may have committed since we opened).
    ///
    /// This follows SQLite's pattern: acquire RESERVED at first write,
    /// re-read metadata to get an authoritative freelist, then proceed.
    /// RESERVED guarantees no other process is writing, so the on-disk
    /// metadata is stable.
    ///
    /// See [[architecture-cross-process-locking]].
    fn begin_write(&mut self) -> Result<()> {
        if self.lock.level() >= crate::storage::lock::LockLevel::Reserved {
            return Ok(()); // Already in write mode
        }

        self.lock.lock_reserved(self.write_timeout)?;

        // Re-read committed metadata from disk. Another process may have
        // committed since we opened, changing the freelist and segments.
        let mut header_buf = [0u8; HEADER_SIZE as usize];
        self.file.read_exact_at(&mut header_buf, 0)?;
        self.header = FileHeader::from_bytes(&header_buf)?;

        let (committed, _used_inactive) = load_metadata(&self.file, &self.header)?;
        self.allocator =
            BlockAllocator::from_state(committed.total_blocks, committed.free_list.clone());
        self.generation = committed
            .segments
            .iter()
            .map(|s| s.generation)
            .max()
            .unwrap_or(0);
        self.committed = committed;

        Ok(())
    }

    /// Write segment data to allocated blocks.
    ///
    /// The segment is not visible to readers until [`commit`](Self::commit)
    /// is called. Data is written to disk immediately so that `commit` only
    /// needs to write metadata and flip the header.
    ///
    /// Automatically acquires RESERVED lock on first write (blocks other
    /// writers). The lock is held until `commit()` completes.
    ///
    /// # Errors
    ///
    /// Returns an error if `data` is empty or the write fails.
    /// Returns `WriterLocked` if another process holds RESERVED.
    pub fn write_segment(&mut self, segment_id: SegmentId, data: &[u8]) -> Result<()> {
        if data.is_empty() {
            return Err(LuciError::InvalidQuery("cannot write empty segment".into()));
        }

        self.begin_write()?;

        let blocks_needed =
            ((data.len() as u64 + BLOCK_SIZE as u64 - 1) / BLOCK_SIZE as u64) as u32;
        let extent = self.allocator.allocate(blocks_needed)?;

        self.file.write_all_at(data, extent.start.byte_offset())?;

        self.pending_segments.push(SegmentEntry::new(
            segment_id,
            extent,
            self.generation + 1,
            data.len() as u64,
        ));

        Ok(())
    }

    /// Read committed segment data by segment ID.
    ///
    /// Only committed segments are visible. Returns `LuciError::IndexNotFound`
    /// if the segment does not exist in the committed state.
    pub fn read_segment(&self, segment_id: SegmentId) -> Result<Vec<u8>> {
        let entry = self
            .committed
            .segments
            .iter()
            .find(|e| e.segment_id == segment_id)
            .ok_or_else(|| LuciError::IndexNotFound(format!("segment {segment_id}")))?;

        let mut buf = vec![0u8; entry.data_len as usize];
        self.file
            .read_exact_at(&mut buf, entry.extent.start.byte_offset())?;
        Ok(buf)
    }

    /// Atomically commit all pending segment writes.
    ///
    /// Implements the six-step atomic commit protocol from [[architecture-storage-format]]:
    ///
    /// 1. Segment data already written to blocks (by `write_segment`)
    /// 2. Serialize new metadata to a freshly allocated block (copy-on-write)
    /// 3. Compute checksum, update the inactive root pointer
    /// 4. `fsync` the data file
    /// 5. Write the 4 KB header with the flipped active root flag
    /// 6. `fsync` the header
    ///
    /// The old inactive root's metadata block is freed and included in the new
    /// free list — but never overwritten until after this commit succeeds, so
    /// both roots remain valid at all times for crash recovery.
    pub fn commit(&mut self) -> Result<()> {
        if self.pending_segments.is_empty()
            && self.pending_removals.is_empty()
            && self.pending_vector_indexes.is_empty()
            && self.pending_vector_index_removals.is_empty()
            && !self.metadata_dirty
        {
            return Ok(());
        }

        // Preserve user_metadata (mapping + deletion bitmap) staged since the
        // last refresh. `begin_write` below re-reads committed state from disk
        // to pick up another process's commits, which would otherwise drop the
        // deletions staged via `set_user_metadata` for a metadata-only commit.
        let staged_user_metadata = self.committed.user_metadata.clone();

        // Acquire RESERVED and refresh allocator + header + committed from disk
        // BEFORE allocating the metadata block or flipping the header. On the
        // normal (write_segment) path this early-returns. On a metadata-only
        // commit (a deletion with no buffered docs) this is the only
        // `begin_write`, and it MUST run first: `begin_write` re-reads the
        // header, so running it after the header flip below would discard the
        // flip and allocate the meta block from a stale allocator — silently
        // reverting the commit on reopen. See
        // luci-index/tests/deletion_persistence.rs.
        self.begin_write()?;
        self.committed.user_metadata = staged_user_metadata;

        self.metadata_dirty = false;

        self.generation += 1;

        // Free the old inactive root's metadata block. This is safe because
        // the active root is our fallback — we never touch its block.
        let old_inactive_meta = self.header.inactive_root_pointer().block_id;
        if let Some(block_id) = old_inactive_meta {
            self.allocator.free(Extent::new(block_id, 1));
        }

        // Allocate a fresh block for the new metadata.
        let meta_extent = self.allocator.allocate(1)?;
        let meta_block = meta_extent.start;

        // Build the new snapshot: committed + pending - removals.
        let mut segments = self.committed.segments.clone();
        segments.extend(self.pending_segments.drain(..));

        // Remove merged source segments and free their blocks.
        if !self.pending_removals.is_empty() {
            segments.retain(|entry| {
                if self.pending_removals.contains(&entry.segment_id) {
                    self.allocator.free(entry.extent);
                    false
                } else {
                    true
                }
            });
            self.pending_removals.clear();
        }

        // Apply vector-index writes + removals. A new write for a
        // field that already has a committed extent replaces it: free
        // the old extent, drop the old entry, then push the new entry.
        // The new entry's blocks were already allocated and written in
        // `write_vector_index` (the same write-then-flip-metadata
        // pattern segments use).
        let mut vector_indexes = self.committed.vector_indexes.clone();
        if !self.pending_vector_index_removals.is_empty() {
            vector_indexes.retain(|entry| {
                if self.pending_vector_index_removals.contains(&entry.field_id) {
                    self.allocator.free(entry.extent);
                    false
                } else {
                    true
                }
            });
            self.pending_vector_index_removals.clear();
        }
        for pending in self.pending_vector_indexes.drain(..) {
            if let Some(pos) = vector_indexes
                .iter()
                .position(|e| e.field_id == pending.field_id)
            {
                self.allocator.free(vector_indexes[pos].extent);
                vector_indexes.remove(pos);
            }
            vector_indexes.push(pending);
        }

        let snapshot = MetadataSnapshot {
            segments,
            vector_indexes,
            total_blocks: self.allocator.total_blocks(),
            free_list: self.allocator.free_list().to_vec(),
            user_metadata: self.committed.user_metadata.clone(),
        };

        assert!(
            snapshot.fits_in_single_block(),
            "metadata overflow chaining not yet implemented"
        );

        // Serialize metadata into a full block (zero-padded for checksum).
        let meta_bytes = snapshot.to_bytes();
        let mut block_buf = vec![0u8; BLOCK_SIZE as usize];
        block_buf[..meta_bytes.len()].copy_from_slice(&meta_bytes);

        // Step 2: write metadata block.
        self.file
            .write_all_at(&block_buf, meta_block.byte_offset())?;

        // Step 3: compute checksum, update header.
        let checksum = xxh3_checksum(&block_buf);
        self.header.commit(meta_block, checksum);

        // fsync data (potentially slow — readers not blocked, we're in RESERVED).
        self.file.sync_all()?;

        // Escalate to EXCLUSIVE (brief — blocks readers for header flip only).
        self.lock.lock_exclusive()?;

        // Write header with flipped active root.
        self.file.write_all_at(&self.header.to_bytes(), 0)?;

        // fsync header (brief).
        self.file.sync_all()?;

        // Downgrade to SHARED (readers unblocked, other writers can proceed).
        self.lock.downgrade_to_shared()?;

        // Update in-memory committed state.
        self.committed = snapshot;

        Ok(())
    }

    /// The currently committed segment entries.
    pub fn segments(&self) -> &[SegmentEntry] {
        &self.committed.segments
    }

    /// The current commit generation.
    pub fn generation(&self) -> u64 {
        self.generation
    }

    /// Set opaque user metadata to be persisted on the next commit.
    pub fn set_user_metadata(&mut self, metadata: Vec<u8>) {
        self.committed.user_metadata = metadata;
        self.metadata_dirty = true;
    }

    /// Get the persisted user metadata (empty if none).
    pub fn user_metadata(&self) -> &[u8] {
        &self.committed.user_metadata
    }

    /// Total number of free (reusable) blocks tracked by the allocator.
    ///
    /// Exposed for integration testing (free-list reclamation validation).
    pub fn free_block_count(&self) -> u64 {
        self.allocator.free_block_count()
    }

    /// Total number of data blocks the file spans.
    ///
    /// Exposed for integration testing.
    pub fn total_blocks(&self) -> u64 {
        self.allocator.total_blocks()
    }

    /// Mark segments for removal on the next commit.
    pub fn remove_segments(&mut self, segment_ids: &[SegmentId]) {
        self.pending_removals.extend_from_slice(segment_ids);
    }

    /// Write a per-field vector index. The bytes are written to a fresh
    /// extent immediately; the entry becomes visible to readers on the
    /// next `commit()`. If the field already had a committed entry, its
    /// old extent is freed during commit.
    pub fn write_vector_index(&mut self, field_id: FieldId, data: &[u8]) -> Result<()> {
        if data.is_empty() {
            return Err(LuciError::InvalidQuery(
                "cannot write empty vector index".into(),
            ));
        }

        self.begin_write()?;

        let blocks_needed =
            ((data.len() as u64 + BLOCK_SIZE as u64 - 1) / BLOCK_SIZE as u64) as u32;
        let extent = self.allocator.allocate(blocks_needed)?;

        self.file.write_all_at(data, extent.start.byte_offset())?;

        // Replace any earlier pending write for the same field so the
        // commit-phase replace logic sees only the latest version.
        self.pending_vector_indexes
            .retain(|e| e.field_id != field_id);
        self.pending_vector_indexes.push(VectorIndexEntry::new(
            field_id,
            extent,
            data.len() as u64,
        ));

        Ok(())
    }

    /// Read committed vector-index bytes for `field_id`. Returns `None`
    /// if no committed index exists for that field.
    pub fn read_vector_index(&self, field_id: FieldId) -> Result<Option<Vec<u8>>> {
        let entry = match self
            .committed
            .vector_indexes
            .iter()
            .find(|e| e.field_id == field_id)
        {
            Some(e) => e,
            None => return Ok(None),
        };

        let mut buf = vec![0u8; entry.data_len as usize];
        self.file
            .read_exact_at(&mut buf, entry.extent.start.byte_offset())?;
        Ok(Some(buf))
    }

    /// List the fields that have a committed vector index.
    pub fn vector_index_fields(&self) -> Vec<FieldId> {
        self.committed
            .vector_indexes
            .iter()
            .map(|e| e.field_id)
            .collect()
    }

    /// Mark the vector index for `field_id` for removal on next commit.
    pub fn remove_vector_index(&mut self, field_id: FieldId) {
        self.pending_vector_index_removals.push(field_id);
    }
}

#[cfg(unix)]
impl crate::storage::Storage for SingleFileDirectory {
    fn write_segment(&mut self, segment_id: SegmentId, data: &[u8]) -> Result<()> {
        self.write_segment(segment_id, data)
    }
    fn read_segment(&self, segment_id: SegmentId) -> Result<Vec<u8>> {
        self.read_segment(segment_id)
    }
    fn commit(&mut self) -> Result<()> {
        self.commit()
    }
    fn segments(&self) -> &[SegmentEntry] {
        self.segments()
    }
    fn generation(&self) -> u64 {
        self.generation()
    }
    fn set_user_metadata(&mut self, metadata: Vec<u8>) {
        self.set_user_metadata(metadata)
    }
    fn user_metadata(&self) -> &[u8] {
        self.user_metadata()
    }
    fn remove_segments(&mut self, segment_ids: &[SegmentId]) {
        self.remove_segments(segment_ids)
    }
    fn write_vector_index(&mut self, field_id: FieldId, data: &[u8]) -> Result<()> {
        self.write_vector_index(field_id, data)
    }
    fn read_vector_index(&self, field_id: FieldId) -> Result<Option<Vec<u8>>> {
        self.read_vector_index(field_id)
    }
    fn vector_index_fields(&self) -> Vec<FieldId> {
        self.vector_index_fields()
    }
    fn remove_vector_index(&mut self, field_id: FieldId) {
        self.remove_vector_index(field_id)
    }
    fn set_write_timeout(&mut self, timeout: std::time::Duration) {
        self.set_write_timeout(timeout)
    }
}

/// Load metadata from the best available root pointer.
///
/// Returns `(snapshot, used_inactive)` where `used_inactive` is true if the
/// active root was invalid and we fell back to the inactive root.
#[cfg(unix)]
fn load_metadata(file: &File, header: &FileHeader) -> Result<(MetadataSnapshot, bool)> {
    // Try active root.
    if let Some(snap) = try_load_root(file, header.active_root_pointer())? {
        return Ok((snap, false));
    }

    // Try inactive root.
    if let Some(snap) = try_load_root(file, header.inactive_root_pointer())? {
        return Ok((snap, true));
    }

    // Both roots empty → fresh (never committed) file.
    if !header.active_root_pointer().is_populated()
        && !header.inactive_root_pointer().is_populated()
    {
        return Ok((MetadataSnapshot::empty(), false));
    }

    // At least one root was populated but both failed validation.
    Err(LuciError::IndexCorrupted(
        "both root pointers failed checksum validation".into(),
    ))
}

/// Try to load metadata from a single root pointer.
///
/// Returns `None` if the root is empty or its checksum doesn't match.
#[cfg(unix)]
fn try_load_root(file: &File, root: &RootPointer) -> Result<Option<MetadataSnapshot>> {
    let block_id = match root.block_id {
        Some(id) => id,
        None => return Ok(None),
    };

    let mut block_buf = vec![0u8; BLOCK_SIZE as usize];
    file.read_exact_at(&mut block_buf, block_id.byte_offset())?;

    let computed = xxh3_checksum(&block_buf);
    if computed != root.checksum {
        return Ok(None);
    }

    let snap = MetadataSnapshot::from_bytes(&block_buf)?;
    Ok(Some(snap))
}

#[cfg(test)]
#[cfg(unix)]
mod tests {
    use super::*;
    use crate::storage::header::FORMAT_VERSION;
    use std::fs;

    /// Create a temp directory for test files. Returns the path.
    fn test_dir() -> std::path::PathBuf {
        let dir = std::env::temp_dir().join(format!("luci_test_{}", std::process::id()));
        fs::create_dir_all(&dir).unwrap();
        dir
    }

    fn test_path(name: &str) -> std::path::PathBuf {
        test_dir().join(name)
    }

    #[test]
    fn create_new_file() {
        let path = test_path("create_new.luci");
        let _ = fs::remove_file(&path);

        let dir = SingleFileDirectory::create(&path).unwrap();
        assert!(dir.segments().is_empty());
        assert_eq!(dir.generation(), 0);

        // File should exist with at least the header.
        let meta = fs::metadata(&path).unwrap();
        assert!(meta.len() >= HEADER_SIZE as u64);

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn create_fails_if_exists() {
        let path = test_path("create_exists.luci");
        let _ = fs::remove_file(&path);

        let _dir = SingleFileDirectory::create(&path).unwrap();
        let err = SingleFileDirectory::create(&path);
        assert!(err.is_err());

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn write_commit_read() {
        let path = test_path("write_commit_read.luci");
        let _ = fs::remove_file(&path);

        let mut dir = SingleFileDirectory::create(&path).unwrap();

        let data = b"hello luci segment data!";
        dir.write_segment(SegmentId::new(1), data).unwrap();
        dir.commit().unwrap();

        let read_back = dir.read_segment(SegmentId::new(1)).unwrap();
        assert_eq!(read_back, data);
        assert_eq!(dir.segments().len(), 1);
        assert_eq!(dir.generation(), 1);

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn multiple_segments() {
        let path = test_path("multiple_segments.luci");
        let _ = fs::remove_file(&path);

        let mut dir = SingleFileDirectory::create(&path).unwrap();

        dir.write_segment(SegmentId::new(1), b"segment-one")
            .unwrap();
        dir.write_segment(SegmentId::new(2), b"segment-two-longer")
            .unwrap();
        dir.commit().unwrap();

        assert_eq!(dir.segments().len(), 2);
        assert_eq!(dir.read_segment(SegmentId::new(1)).unwrap(), b"segment-one");
        assert_eq!(
            dir.read_segment(SegmentId::new(2)).unwrap(),
            b"segment-two-longer"
        );

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn reopen_after_commit() {
        let path = test_path("reopen.luci");
        let _ = fs::remove_file(&path);

        // Create and write.
        {
            let mut dir = SingleFileDirectory::create(&path).unwrap();
            dir.write_segment(SegmentId::new(1), b"persistent data")
                .unwrap();
            dir.commit().unwrap();
        }

        // Reopen and read.
        {
            let dir = SingleFileDirectory::open(&path).unwrap();
            assert_eq!(dir.segments().len(), 1);
            assert_eq!(
                dir.read_segment(SegmentId::new(1)).unwrap(),
                b"persistent data"
            );
            assert_eq!(dir.generation(), 1);
        }

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn uncommitted_writes_not_visible_after_reopen() {
        let path = test_path("uncommitted.luci");
        let _ = fs::remove_file(&path);

        {
            let mut dir = SingleFileDirectory::create(&path).unwrap();
            dir.write_segment(SegmentId::new(1), b"committed").unwrap();
            dir.commit().unwrap();

            // Write but don't commit.
            dir.write_segment(SegmentId::new(2), b"uncommitted")
                .unwrap();
        }

        {
            let dir = SingleFileDirectory::open(&path).unwrap();
            assert_eq!(dir.segments().len(), 1);
            assert_eq!(dir.read_segment(SegmentId::new(1)).unwrap(), b"committed");
            assert!(dir.read_segment(SegmentId::new(2)).is_err());
        }

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn multiple_commits() {
        let path = test_path("multi_commit.luci");
        let _ = fs::remove_file(&path);

        let mut dir = SingleFileDirectory::create(&path).unwrap();

        dir.write_segment(SegmentId::new(1), b"first").unwrap();
        dir.commit().unwrap();
        assert_eq!(dir.generation(), 1);

        dir.write_segment(SegmentId::new(2), b"second").unwrap();
        dir.commit().unwrap();
        assert_eq!(dir.generation(), 2);

        dir.write_segment(SegmentId::new(3), b"third").unwrap();
        dir.commit().unwrap();
        assert_eq!(dir.generation(), 3);

        assert_eq!(dir.segments().len(), 3);
        assert_eq!(dir.read_segment(SegmentId::new(1)).unwrap(), b"first");
        assert_eq!(dir.read_segment(SegmentId::new(2)).unwrap(), b"second");
        assert_eq!(dir.read_segment(SegmentId::new(3)).unwrap(), b"third");

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn reopen_after_multiple_commits() {
        let path = test_path("reopen_multi.luci");
        let _ = fs::remove_file(&path);

        {
            let mut dir = SingleFileDirectory::create(&path).unwrap();
            dir.write_segment(SegmentId::new(1), b"aaa").unwrap();
            dir.commit().unwrap();
            dir.write_segment(SegmentId::new(2), b"bbb").unwrap();
            dir.commit().unwrap();
            dir.write_segment(SegmentId::new(3), b"ccc").unwrap();
            dir.commit().unwrap();
        }

        {
            let dir = SingleFileDirectory::open(&path).unwrap();
            assert_eq!(dir.segments().len(), 3);
            assert_eq!(dir.read_segment(SegmentId::new(1)).unwrap(), b"aaa");
            assert_eq!(dir.read_segment(SegmentId::new(2)).unwrap(), b"bbb");
            assert_eq!(dir.read_segment(SegmentId::new(3)).unwrap(), b"ccc");
            assert_eq!(dir.generation(), 3);
        }

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn large_segment_spanning_multiple_blocks() {
        let path = test_path("large_segment.luci");
        let _ = fs::remove_file(&path);

        let mut dir = SingleFileDirectory::create(&path).unwrap();

        // Write data larger than one block (256 KB).
        let data = vec![0xABu8; BLOCK_SIZE as usize * 3 + 1000];
        dir.write_segment(SegmentId::new(1), &data).unwrap();
        dir.commit().unwrap();

        let read_back = dir.read_segment(SegmentId::new(1)).unwrap();
        assert_eq!(read_back, data);

        // Verify it allocated 4 blocks (3 full + 1 partial).
        assert_eq!(dir.segments()[0].extent.count, 4);

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn read_nonexistent_segment_is_error() {
        let path = test_path("read_nonexist.luci");
        let _ = fs::remove_file(&path);

        let dir = SingleFileDirectory::create(&path).unwrap();
        let err = dir.read_segment(SegmentId::new(999));
        assert!(err.is_err());

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn empty_commit_is_noop() {
        let path = test_path("empty_commit.luci");
        let _ = fs::remove_file(&path);

        let mut dir = SingleFileDirectory::create(&path).unwrap();
        // Committing with no pending segments should be a no-op.
        dir.commit().unwrap();
        assert_eq!(dir.generation(), 0);
        assert!(dir.segments().is_empty());

        fs::remove_file(&path).unwrap();
    }

    #[test]
    fn simulated_torn_header_falls_back() {
        let path = test_path("torn_header.luci");
        let _ = fs::remove_file(&path);

        // Create file with two commits so both roots are populated.
        {
            let mut dir = SingleFileDirectory::create(&path).unwrap();
            dir.write_segment(SegmentId::new(1), b"first-commit")
                .unwrap();
            dir.commit().unwrap();
            dir.write_segment(SegmentId::new(2), b"second-commit")
                .unwrap();
            dir.commit().unwrap();
        }

        // Corrupt the active root's checksum in the header to simulate a
        // torn header write.
        {
            let file = OpenOptions::new()
                .read(true)
                .write(true)
                .open(&path)
                .unwrap();
            let mut header_buf = [0u8; HEADER_SIZE as usize];
            file.read_exact_at(&mut header_buf, 0).unwrap();
            let header = FileHeader::from_bytes(&header_buf).unwrap();

            // Corrupt the active root's checksum.
            let checksum_offset = match header.active_root {
                crate::storage::ActiveRoot::A => 24usize, // OFF_ROOT_A_CHECKSUM
                crate::storage::ActiveRoot::B => 40usize, // OFF_ROOT_B_CHECKSUM
            };
            header_buf[checksum_offset] ^= 0xFF;
            file.write_all_at(&header_buf, 0).unwrap();
            file.sync_all().unwrap();
        }

        // Open should recover using the inactive root (first commit).
        {
            let dir = SingleFileDirectory::open(&path).unwrap();
            assert_eq!(dir.segments().len(), 1);
            assert_eq!(
                dir.read_segment(SegmentId::new(1)).unwrap(),
                b"first-commit"
            );
            // Second commit's segment should not be visible.
            assert!(dir.read_segment(SegmentId::new(2)).is_err());
        }

        fs::remove_file(&path).unwrap();
    }

    /// Test 11: a v2 file committed by a new binary is re-stamped to the current
    /// `FORMAT_VERSION` on ANY write — here a metadata-only commit — so an old
    /// binary then rejects it loudly rather than silently misreading a v3
    /// `KeywordBlocked` column. Guards the [[code-must-not-lie]]
    /// silent-`_id`-drop hole. The stamp survives only because `commit()` is
    /// refresh-first. See [[optimization-keyword-dict-offset-index]].
    #[test]
    fn version_stamped_on_any_commit() {
        let path = test_path("version_stamp.luci");
        let _ = fs::remove_file(&path);

        // Create a normal file, then overwrite its header with a fabricated v2
        // one (mirroring create()'s own write_all_at) to simulate an old binary.
        SingleFileDirectory::create(&path).unwrap();
        {
            let file = OpenOptions::new()
                .read(true)
                .write(true)
                .open(&path)
                .unwrap();
            let v2 = FileHeader::with_format_version(2).to_bytes();
            file.write_all_at(&v2, 0).unwrap();
            file.sync_all().unwrap();
        }
        // Sanity: the on-disk version is now 2.
        {
            let file = OpenOptions::new().read(true).open(&path).unwrap();
            let mut buf = [0u8; HEADER_SIZE as usize];
            file.read_exact_at(&mut buf, 0).unwrap();
            assert_eq!(FileHeader::from_bytes(&buf).unwrap().format_version, 2);
        }

        // The new binary opens the v2 file and does a metadata-only commit
        // (set_user_metadata marks the directory dirty with no segments — the
        // same shape as a bare deletion commit).
        {
            let mut dir = SingleFileDirectory::open(&path).unwrap();
            dir.set_user_metadata(b"deletion-marker".to_vec());
            dir.commit().unwrap();
        }

        // commit() re-stamped the header to the writing binary's FORMAT_VERSION.
        // The complementary "old binary rejects a v3 file" half is covered by
        // header.rs::future_version_is_rejected.
        {
            let file = OpenOptions::new().read(true).open(&path).unwrap();
            let mut buf = [0u8; HEADER_SIZE as usize];
            file.read_exact_at(&mut buf, 0).unwrap();
            assert_eq!(
                FileHeader::from_bytes(&buf).unwrap().format_version,
                FORMAT_VERSION
            );
        }

        fs::remove_file(&path).unwrap();
    }
}