laminar_storage/per_core_wal/

writer.rs

//! Per-core WAL writer for lock-free segment writes.
//!
//! Supports two write paths:
//! - **Direct** (`append` / `sync`): uses `BufWriter<File>` — always available.
//! - **Storage I/O** (`append_via_storage_io` / `sync_via_storage_io`): uses the
//!   [`StorageIo`] trait for non-blocking writes. On Linux with `io_uring` +
//!   SQPOLL this is zero-syscall. On Windows it falls back to synchronous I/O
//!   through the same trait interface.

use std::fs::{File, OpenOptions};
use std::io::{BufWriter, Write};
use std::path::{Path, PathBuf};
use std::time::Instant;

use laminar_core::storage_io::{IoCompletion, IoFd, StorageIo, StorageIoError};
use rkyv::api::high;
use rkyv::rancor::Error as RkyvError;
use rkyv::util::AlignedVec;

use super::entry::PerCoreWalEntry;
use super::error::PerCoreWalError;

/// Size of the record header (length + CRC32).
const RECORD_HEADER_SIZE: u64 = 8;

/// Per-core WAL writer.
///
/// Each core owns its own writer, eliminating cross-core synchronization on the write path.
/// Record format is compatible with `[length: 4][crc32: 4][data: length]`
pub struct CoreWalWriter {
    /// Core ID this writer belongs to.
    core_id: usize,
    /// Buffered writer for efficient writes.
    writer: BufWriter<File>,
    /// Path to the segment file.
    path: PathBuf,
    /// Current write position in bytes (includes buffered, un-synced data).
    position: u64,
    /// Last synced position (data confirmed durable via `fdatasync`).
    ///
    /// Only this position is safe for checkpoint manifests — data beyond
    /// it may be lost on crash.
    synced_position: u64,
    /// Current epoch (set by manager during checkpoint).
    epoch: u64,
    /// Core-local sequence number (monotonically increasing).
    sequence: u64,
    /// Last sync time for group commit.
    last_sync: Instant,
    /// Number of entries since last sync.
    entries_since_sync: u64,
    /// Pre-allocated write buffer reused across `append()` calls.
    /// Grows to high-water mark and stays, eliminating per-append allocation.
    write_buffer: Vec<u8>,
    /// Reusable rkyv serialization buffer (avoids `AlignedVec` alloc per append).
    serialize_buffer: AlignedVec,
    /// Pending sync state: (token, position boundary at submission time).
    /// The boundary is the `position` when `sync_via_storage_io` was called.
    /// On completion, `synced_position` advances to the boundary — not to
    /// the current `position`, which may have advanced from later appends.
    pending_sync: Option<PendingSync>,
}

/// In-flight fdatasync: token + position at submission time.
#[derive(Debug, Clone, Copy)]
struct PendingSync {
    token: u64,
    /// Only data up to this offset is durable when the CQE arrives.
    boundary: u64,
}

impl CoreWalWriter {
    /// Creates a new per-core WAL writer.
    ///
    /// # Arguments
    ///
    /// * `core_id` - The core ID for this writer
    /// * `path` - Path to the segment file
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be created or opened.
    pub fn new(core_id: usize, path: &Path) -> Result<Self, PerCoreWalError> {
        let file = OpenOptions::new().create(true).append(true).open(path)?;

        let position = file.metadata()?.len();

        Ok(Self {
            core_id,
            writer: BufWriter::with_capacity(64 * 1024, file), // 64KB buffer
            path: path.to_path_buf(),
            position,
            synced_position: position,
            epoch: 0,
            sequence: 0,
            last_sync: Instant::now(),
            entries_since_sync: 0,
            write_buffer: Vec::with_capacity(4096),
            serialize_buffer: AlignedVec::with_capacity(256),
            pending_sync: None,
        })
    }

    /// Opens an existing segment file at a specific position.
    ///
    /// Used during recovery to resume writing.
    ///
    /// # Errors
    ///
    /// Returns an error if the file cannot be opened or truncated.
    pub fn open_at(core_id: usize, path: &Path, position: u64) -> Result<Self, PerCoreWalError> {
        let file = OpenOptions::new().write(true).open(path)?;

        // Truncate to the specified position (in case of torn writes)
        file.set_len(position)?;

        let file = OpenOptions::new().append(true).open(path)?;

        Ok(Self {
            core_id,
            writer: BufWriter::with_capacity(64 * 1024, file),
            path: path.to_path_buf(),
            position,
            synced_position: position,
            epoch: 0,
            sequence: 0,
            last_sync: Instant::now(),
            entries_since_sync: 0,
            write_buffer: Vec::with_capacity(4096),
            serialize_buffer: AlignedVec::with_capacity(256),
            pending_sync: None,
        })
    }

    /// Returns the core ID for this writer.
    #[must_use]
    pub fn core_id(&self) -> usize {
        self.core_id
    }

    /// Returns the current write position in bytes (includes un-synced data).
    #[must_use]
    pub fn position(&self) -> u64 {
        self.position
    }

    /// Returns the last synced position (durable after `fdatasync`).
    ///
    /// Only data up to this position is guaranteed to survive a crash.
    /// Use this for checkpoint manifests instead of [`position()`](Self::position).
    #[must_use]
    pub fn synced_position(&self) -> u64 {
        self.synced_position
    }

    /// Returns the current epoch.
    #[must_use]
    pub fn epoch(&self) -> u64 {
        self.epoch
    }

    /// Returns the current sequence number.
    #[must_use]
    pub fn sequence(&self) -> u64 {
        self.sequence
    }

    /// Returns the path to the segment file.
    #[must_use]
    pub fn path(&self) -> &Path {
        &self.path
    }

    /// Returns the number of entries since last sync.
    #[must_use]
    pub fn entries_since_sync(&self) -> u64 {
        self.entries_since_sync
    }

    /// Sets the current epoch (called by manager during checkpoint).
    pub fn set_epoch(&mut self, epoch: u64) {
        self.epoch = epoch;
    }

    /// Appends a Put operation to the WAL.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O fails.
    #[inline]
    #[allow(clippy::cast_possible_truncation)] // core_id bounded by physical CPU count (< u16::MAX)
    pub fn append_put(&mut self, key: &[u8], value: &[u8]) -> Result<u64, PerCoreWalError> {
        let ts = PerCoreWalEntry::now_ns();
        let entry = PerCoreWalEntry::put(
            self.core_id as u16,
            self.epoch,
            self.sequence,
            key.to_vec(),
            value.to_vec(),
            ts,
        );
        self.append(&entry)
    }

    /// Appends a Delete operation to the WAL.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O fails.
    #[inline]
    #[allow(clippy::cast_possible_truncation)] // core_id bounded by physical CPU count (< u16::MAX)
    pub fn append_delete(&mut self, key: &[u8]) -> Result<u64, PerCoreWalError> {
        let ts = PerCoreWalEntry::now_ns();
        let entry = PerCoreWalEntry::delete(
            self.core_id as u16,
            self.epoch,
            self.sequence,
            key.to_vec(),
            ts,
        );
        self.append(&entry)
    }

    /// Appends a Put operation via [`StorageIo`].
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O submission fails.
    #[inline]
    #[allow(clippy::cast_possible_truncation)]
    pub fn append_put_via_storage_io(
        &mut self,
        key: &[u8],
        value: &[u8],
        sio: &mut dyn StorageIo,
        fd: IoFd,
    ) -> Result<u64, PerCoreWalError> {
        let ts = PerCoreWalEntry::now_ns();
        let entry = PerCoreWalEntry::put(
            self.core_id as u16,
            self.epoch,
            self.sequence,
            key.to_vec(),
            value.to_vec(),
            ts,
        );
        self.append_via_storage_io(&entry, sio, fd)
    }

    /// Appends a Delete operation via [`StorageIo`].
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O submission fails.
    #[inline]
    #[allow(clippy::cast_possible_truncation)]
    pub fn append_delete_via_storage_io(
        &mut self,
        key: &[u8],
        sio: &mut dyn StorageIo,
        fd: IoFd,
    ) -> Result<u64, PerCoreWalError> {
        let ts = PerCoreWalEntry::now_ns();
        let entry = PerCoreWalEntry::delete(
            self.core_id as u16,
            self.epoch,
            self.sequence,
            key.to_vec(),
            ts,
        );
        self.append_via_storage_io(&entry, sio, fd)
    }

    /// Serialize an entry into `self.write_buffer` (header + CRC + data).
    ///
    /// Shared by [`append`] and [`append_via_storage_io`]. After this call,
    /// `self.write_buffer` contains the complete record ready for I/O.
    fn serialize_entry(&mut self, entry: &PerCoreWalEntry) -> Result<(), PerCoreWalError> {
        self.serialize_buffer.clear();
        let taken = std::mem::take(&mut self.serialize_buffer);
        let bytes = high::to_bytes_in::<_, RkyvError>(entry, taken)
            .map_err(|e| PerCoreWalError::Serialization(e.to_string()))?;

        let crc = crc32c::crc32c(&bytes);

        #[allow(clippy::cast_possible_truncation)]
        let len = bytes.len() as u32;
        self.write_buffer.clear();
        #[allow(clippy::cast_possible_truncation)]
        self.write_buffer
            .reserve(RECORD_HEADER_SIZE as usize + bytes.len());
        self.write_buffer.extend_from_slice(&len.to_le_bytes());
        self.write_buffer.extend_from_slice(&crc.to_le_bytes());
        self.write_buffer.extend_from_slice(&bytes);

        self.serialize_buffer = bytes;
        Ok(())
    }

    /// Advance position/sequence counters after a successful write.
    fn advance_position(&mut self) {
        #[allow(clippy::cast_possible_truncation)]
        let data_len = self.write_buffer.len() as u64 - RECORD_HEADER_SIZE;
        self.position += RECORD_HEADER_SIZE + data_len;
        self.sequence += 1;
        self.entries_since_sync += 1;
    }

    /// Appends a raw entry to the WAL.
    ///
    /// Record format: `[length: 4 bytes][crc32: 4 bytes][data: length bytes]`
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O fails.
    pub fn append(&mut self, entry: &PerCoreWalEntry) -> Result<u64, PerCoreWalError> {
        let start_pos = self.position;
        self.serialize_entry(entry)?;
        self.writer.write_all(&self.write_buffer)?;
        self.advance_position();
        Ok(start_pos)
    }

    /// Appends a raw entry to the WAL via a [`StorageIo`] backend.
    ///
    /// Identical to [`append`](Self::append) but uses non-blocking I/O.
    /// The caller must poll `sio.poll_completions()` to drain completions.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization fails or the I/O submission fails.
    pub fn append_via_storage_io(
        &mut self,
        entry: &PerCoreWalEntry,
        sio: &mut dyn StorageIo,
        fd: IoFd,
    ) -> Result<u64, PerCoreWalError> {
        let start_pos = self.position;
        self.serialize_entry(entry)?;
        sio.submit_append(fd, &self.write_buffer)
            .map_err(storage_io_to_wal_error)?;
        self.advance_position();
        Ok(start_pos)
    }

    /// Syncs the WAL segment via a [`StorageIo`] backend.
    ///
    /// Submits an `fdatasync` through the I/O backend. Non-blocking — the
    /// caller must poll for the completion token.
    ///
    /// # Errors
    ///
    /// Returns an error if the submission fails.
    pub fn sync_via_storage_io(
        &mut self,
        sio: &mut dyn StorageIo,
        fd: IoFd,
    ) -> Result<u64, PerCoreWalError> {
        if self.pending_sync.is_some() {
            return Err(PerCoreWalError::Io(std::io::Error::other(
                "sync already in flight — wait for completion before submitting another",
            )));
        }
        let token = sio.submit_datasync(fd).map_err(storage_io_to_wal_error)?;
        self.pending_sync = Some(PendingSync {
            token,
            boundary: self.position,
        });
        Ok(token)
    }

    /// Check I/O completions and update sync state.
    ///
    /// Scans `completions` for the pending sync token. If found and
    /// successful, calls [`mark_synced`](Self::mark_synced). If found
    /// and failed (negative result = errno), returns an error.
    ///
    /// Returns `Ok(true)` if the sync completed successfully,
    /// `Ok(false)` if no matching completion was found,
    /// or `Err` if the sync failed.
    ///
    /// Call this after every `StorageIo::poll_completions` round.
    ///
    /// # Errors
    ///
    /// Returns `PerCoreWalError::Io` if the kernel reported a sync failure.
    pub fn check_completions(
        &mut self,
        completions: &[IoCompletion],
    ) -> Result<bool, PerCoreWalError> {
        let Some(pending) = self.pending_sync else {
            return Ok(false);
        };
        for c in completions {
            if c.token == pending.token {
                self.pending_sync = None;
                if c.result >= 0 {
                    // Advance synced_position only to the boundary that was
                    // current when the sync was submitted — not to the
                    // current position, which may include later appends.
                    self.synced_position = pending.boundary;
                    self.last_sync = Instant::now();
                    self.entries_since_sync = 0;
                    return Ok(true);
                }
                return Err(PerCoreWalError::Io(std::io::Error::from_raw_os_error(
                    -c.result,
                )));
            }
        }
        Ok(false)
    }

    /// Returns `true` if a sync is pending (submitted but not yet completed).
    #[must_use]
    pub fn is_sync_pending(&self) -> bool {
        self.pending_sync.is_some()
    }

    /// Mark the current write position as synced (durable).
    ///
    /// Used by the direct (non-`StorageIo`) sync path after a blocking
    /// `fdatasync` where all data up to `position` is guaranteed durable.
    /// The `StorageIo` path uses [`Self::check_completions`] instead, which
    /// advances only to the boundary recorded at submission time.
    pub fn mark_synced(&mut self) {
        self.synced_position = self.position;
        self.last_sync = Instant::now();
        self.entries_since_sync = 0;
    }

    /// Appends a Checkpoint marker to the WAL.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O fails.
    #[allow(clippy::cast_possible_truncation)] // core_id bounded by physical CPU count (< u16::MAX)
    pub fn append_checkpoint(&mut self, checkpoint_id: u64) -> Result<u64, PerCoreWalError> {
        let ts = PerCoreWalEntry::now_ns();
        let entry = PerCoreWalEntry::checkpoint(
            self.core_id as u16,
            self.epoch,
            self.sequence,
            checkpoint_id,
            ts,
        );
        self.append(&entry)
    }

    /// Appends an `EpochBarrier` to the WAL.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O fails.
    #[allow(clippy::cast_possible_truncation)] // core_id bounded by physical CPU count (< u16::MAX)
    pub fn append_epoch_barrier(&mut self) -> Result<u64, PerCoreWalError> {
        let ts = PerCoreWalEntry::now_ns();
        let entry =
            PerCoreWalEntry::epoch_barrier(self.core_id as u16, self.epoch, self.sequence, ts);
        self.append(&entry)
    }

    /// Appends a Commit entry to the WAL.
    ///
    /// # Errors
    ///
    /// Returns an error if serialization or I/O fails.
    #[allow(clippy::cast_possible_truncation)] // core_id bounded by physical CPU count (< u16::MAX)
    #[allow(clippy::disallowed_types)] // cold path: WAL coordination
    pub fn append_commit(
        &mut self,
        offsets: std::collections::HashMap<String, u64>,
        watermark: Option<i64>,
    ) -> Result<u64, PerCoreWalError> {
        let ts = PerCoreWalEntry::now_ns();
        let entry = PerCoreWalEntry::commit(
            self.core_id as u16,
            self.epoch,
            self.sequence,
            offsets,
            watermark,
            ts,
        );
        self.append(&entry)
    }

    /// Syncs the WAL segment to disk using fdatasync.
    ///
    /// Uses `sync_data()` instead of `sync_all()` for better performance.
    ///
    /// # Errors
    ///
    /// Returns an error if the sync fails.
    pub fn sync(&mut self) -> Result<(), PerCoreWalError> {
        self.writer.flush()?;
        // Use sync_data() (fdatasync) instead of sync_all() (fsync)
        self.writer.get_ref().sync_data()?;
        self.synced_position = self.position;
        self.last_sync = Instant::now();
        self.entries_since_sync = 0;
        Ok(())
    }

    /// Truncates the segment file at the specified position.
    ///
    /// Used after checkpoint to remove entries that have been checkpointed.
    ///
    /// # Errors
    ///
    /// Returns an error if truncation fails.
    pub fn truncate(&mut self, position: u64) -> Result<(), PerCoreWalError> {
        self.sync()?;

        // Close current writer by dropping, then truncate
        let file = OpenOptions::new()
            .write(true)
            .truncate(false)
            .open(&self.path)?;

        // Sync after truncation to make it durable.
        // Without this, a crash could leave the file at its old length.
        file.set_len(position)?;
        file.sync_all()?;

        // Reopen for append
        let file = OpenOptions::new().append(true).open(&self.path)?;

        self.writer = BufWriter::with_capacity(64 * 1024, file);
        self.position = position;
        self.synced_position = position;

        Ok(())
    }

    /// Resets the segment (truncates to zero).
    ///
    /// Used after successful checkpoint to clear the WAL.
    ///
    /// # Errors
    ///
    /// Returns an error if truncation fails.
    pub fn reset(&mut self) -> Result<(), PerCoreWalError> {
        self.truncate(0)?;
        self.sequence = 0;
        Ok(())
    }
}

/// Convert a [`StorageIoError`] to a [`PerCoreWalError`].
fn storage_io_to_wal_error(e: StorageIoError) -> PerCoreWalError {
    match e {
        StorageIoError::Io(io_err) => PerCoreWalError::Io(io_err),
        other => PerCoreWalError::Io(std::io::Error::other(other.to_string())),
    }
}

impl std::fmt::Debug for CoreWalWriter {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("CoreWalWriter")
            .field("core_id", &self.core_id)
            .field("path", &self.path)
            .field("position", &self.position)
            .field("synced_position", &self.synced_position)
            .field("epoch", &self.epoch)
            .field("sequence", &self.sequence)
            .field("entries_since_sync", &self.entries_since_sync)
            .finish_non_exhaustive()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tempfile::TempDir;

    fn create_temp_writer(core_id: usize) -> (CoreWalWriter, TempDir) {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join(format!("wal-{core_id}.log"));
        let writer = CoreWalWriter::new(core_id, &path).unwrap();
        (writer, temp_dir)
    }

    #[test]
    fn test_writer_creation() {
        let (writer, _temp_dir) = create_temp_writer(0);
        assert_eq!(writer.core_id(), 0);
        assert_eq!(writer.position(), 0);
        assert_eq!(writer.epoch(), 0);
        assert_eq!(writer.sequence(), 0);
    }

    #[test]
    fn test_append_put() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        let pos = writer.append_put(b"key1", b"value1").unwrap();
        assert_eq!(pos, 0);
        assert!(writer.position() > 0);
        assert_eq!(writer.sequence(), 1);

        let pos2 = writer.append_put(b"key2", b"value2").unwrap();
        assert!(pos2 > pos);
        assert_eq!(writer.sequence(), 2);
    }

    #[test]
    fn test_append_delete() {
        let (mut writer, _temp_dir) = create_temp_writer(1);

        let pos = writer.append_delete(b"key1").unwrap();
        assert_eq!(pos, 0);
        assert!(writer.position() > 0);
    }

    #[test]
    fn test_sync() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        writer.append_put(b"key1", b"value1").unwrap();
        assert_eq!(writer.entries_since_sync(), 1);

        writer.sync().unwrap();
        assert_eq!(writer.entries_since_sync(), 0);
    }

    #[test]
    fn test_epoch_setting() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        assert_eq!(writer.epoch(), 0);
        writer.set_epoch(5);
        assert_eq!(writer.epoch(), 5);
    }

    #[test]
    fn test_truncate() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        writer.append_put(b"key1", b"value1").unwrap();
        let pos1 = writer.position();

        writer.append_put(b"key2", b"value2").unwrap();
        let pos2 = writer.position();

        assert!(pos2 > pos1);

        writer.truncate(pos1).unwrap();
        assert_eq!(writer.position(), pos1);
    }

    #[test]
    fn test_reset() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        writer.append_put(b"key1", b"value1").unwrap();
        writer.append_put(b"key2", b"value2").unwrap();
        assert!(writer.position() > 0);
        assert_eq!(writer.sequence(), 2);

        writer.reset().unwrap();
        assert_eq!(writer.position(), 0);
        assert_eq!(writer.sequence(), 0);
    }

    #[test]
    fn test_append_checkpoint() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        let pos = writer.append_checkpoint(100).unwrap();
        assert_eq!(pos, 0);
        assert!(writer.position() > 0);
    }

    #[test]
    fn test_append_epoch_barrier() {
        let (mut writer, _temp_dir) = create_temp_writer(0);
        writer.set_epoch(5);

        let pos = writer.append_epoch_barrier().unwrap();
        assert_eq!(pos, 0);
        assert!(writer.position() > 0);
    }

    #[test]
    fn test_append_commit() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        #[allow(clippy::disallowed_types)] // cold path: WAL coordination
        let mut offsets = std::collections::HashMap::new();
        offsets.insert("topic1".to_string(), 100);

        let pos = writer.append_commit(offsets, Some(12345)).unwrap();
        assert_eq!(pos, 0);
        assert!(writer.position() > 0);
    }

    #[test]
    fn test_synced_position_tracks_sync() {
        let (mut writer, _temp_dir) = create_temp_writer(0);

        // Initially both positions are 0
        assert_eq!(writer.position(), 0);
        assert_eq!(writer.synced_position(), 0);

        // After write, position advances but synced_position stays
        writer.append_put(b"key1", b"value1").unwrap();
        assert!(writer.position() > 0);
        assert_eq!(writer.synced_position(), 0);

        // After sync, synced_position catches up
        let pos_before_sync = writer.position();
        writer.sync().unwrap();
        assert_eq!(writer.synced_position(), pos_before_sync);

        // Another write without sync
        writer.append_put(b"key2", b"value2").unwrap();
        assert!(writer.position() > writer.synced_position());
        assert_eq!(writer.synced_position(), pos_before_sync);
    }

    #[test]
    fn test_debug_format() {
        let (writer, _temp_dir) = create_temp_writer(42);
        let debug_str = format!("{writer:?}");
        assert!(debug_str.contains("CoreWalWriter"));
        assert!(debug_str.contains("42"));
    }

    #[test]
    fn test_open_at() {
        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("wal-0.log");

        // Create and write some data
        {
            let mut writer = CoreWalWriter::new(0, &path).unwrap();
            writer.append_put(b"key1", b"value1").unwrap();
            writer.append_put(b"key2", b"value2").unwrap();
            writer.sync().unwrap();
        }

        // Get file size
        let file_size = std::fs::metadata(&path).unwrap().len();

        // Open at position 0 (truncates everything)
        let writer = CoreWalWriter::open_at(0, &path, 0).unwrap();
        assert_eq!(writer.position(), 0);

        // File should be truncated
        let new_size = std::fs::metadata(&path).unwrap().len();
        assert!(new_size < file_size);
    }

    #[test]
    fn test_append_via_storage_io() {
        use laminar_core::storage_io::SyncStorageIo;

        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("wal-sio.log");

        // Create the writer (owns the BufWriter path)
        let mut writer = CoreWalWriter::new(0, &path).unwrap();

        // Create a SyncStorageIo backend and register the same file
        let mut sio = SyncStorageIo::new();
        let sio_file = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        let fd = sio.register_fd(sio_file).unwrap();

        // Write via storage I/O path
        let pos = writer
            .append_put_via_storage_io(b"key1", b"value1", &mut sio, fd)
            .unwrap();
        assert_eq!(pos, 0);
        assert!(writer.position() > 0);
        assert_eq!(writer.sequence(), 1);

        // Sync via storage I/O path — stores token internally
        assert!(!writer.is_sync_pending());
        let _token = writer.sync_via_storage_io(&mut sio, fd).unwrap();
        assert!(writer.is_sync_pending());
        assert_eq!(writer.synced_position(), 0); // not yet synced

        // Drain completions (write + sync)
        let mut completions = Vec::new();
        sio.poll_completions(&mut completions);
        assert!(completions.len() >= 2);

        // Pass completions to writer — it matches the sync token and calls mark_synced
        let synced = writer.check_completions(&completions).unwrap();
        assert!(synced);
        assert!(!writer.is_sync_pending());
        assert_eq!(writer.synced_position(), writer.position());

        // Verify data is on disk and readable via the standard reader
        let file_size = std::fs::metadata(&path).unwrap().len();
        assert!(file_size > 0);
    }

    #[test]
    fn test_check_completions_no_match() {
        use laminar_core::storage_io::IoCompletion;

        let (mut writer, _temp_dir) = create_temp_writer(0);

        // No pending sync → returns Ok(false)
        let completions = vec![IoCompletion {
            token: 999,
            result: 0,
        }];
        assert!(!writer.check_completions(&completions).unwrap());

        // Empty completions → returns Ok(false)
        assert!(!writer.check_completions(&[]).unwrap());
    }

    #[test]
    fn test_sync_boundary_not_advanced_past_submission() {
        use laminar_core::storage_io::SyncStorageIo;

        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("wal-boundary.log");
        let mut writer = CoreWalWriter::new(0, &path).unwrap();

        let mut sio = SyncStorageIo::new();
        let sio_file = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        let fd = sio.register_fd(sio_file).unwrap();

        // Write, then sync
        writer
            .append_put_via_storage_io(b"k1", b"v1", &mut sio, fd)
            .unwrap();
        let pos_at_sync = writer.position();
        let _token = writer.sync_via_storage_io(&mut sio, fd).unwrap();

        // Write MORE after sync was submitted
        writer
            .append_put_via_storage_io(b"k2", b"v2", &mut sio, fd)
            .unwrap();
        assert!(writer.position() > pos_at_sync);

        // Drain and check — synced_position must be pos_at_sync, NOT position()
        let mut completions = Vec::new();
        sio.poll_completions(&mut completions);
        let synced = writer.check_completions(&completions).unwrap();
        assert!(synced);
        assert_eq!(writer.synced_position(), pos_at_sync);
        assert!(writer.synced_position() < writer.position());
    }

    #[test]
    fn test_overlapping_sync_rejected() {
        use laminar_core::storage_io::SyncStorageIo;

        let temp_dir = TempDir::new().unwrap();
        let path = temp_dir.path().join("wal-overlap.log");
        let mut writer = CoreWalWriter::new(0, &path).unwrap();

        let mut sio = SyncStorageIo::new();
        let sio_file = std::fs::OpenOptions::new()
            .append(true)
            .open(&path)
            .unwrap();
        let fd = sio.register_fd(sio_file).unwrap();

        writer
            .append_put_via_storage_io(b"k1", b"v1", &mut sio, fd)
            .unwrap();
        writer.sync_via_storage_io(&mut sio, fd).unwrap();

        // Second sync while first is pending → error
        let result = writer.sync_via_storage_io(&mut sio, fd);
        assert!(result.is_err());
    }
}