powdb_storage/

wal.rs

use crate::error::StorageError;
use std::fs::{File, OpenOptions};
use std::io::{self, BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::JoinHandle;
use std::time::Duration;
use tracing::debug;

/// Process-wide WAL fsync accounting.
///
/// PowDB is a single-writer, fsync-bound engine: how long an fsync takes, and
/// how many are issued, is the single most useful signal an operator has for
/// write-path health. These are plain relaxed atomics updated at the two
/// places that call `sync_data` (the group-commit leader and the Normal-mode
/// background flusher), so a reader (the server's `/metrics` endpoint) can
/// sample them without taking any lock, including the engine lock.
///
/// Process-global rather than per-`Wal` because a PowDB server process serves
/// exactly one data directory, and the metrics endpoint must not reach through
/// the engine `RwLock` to read them.
static FSYNC_TOTAL: AtomicU64 = AtomicU64::new(0);
static FSYNC_NANOS_TOTAL: AtomicU64 = AtomicU64::new(0);
static FSYNC_FAILURES_TOTAL: AtomicU64 = AtomicU64::new(0);

/// Snapshot of the process-wide WAL fsync counters.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub struct WalFsyncStats {
    /// Successful `sync_data` calls issued against a WAL file.
    pub count: u64,
    /// Total nanoseconds spent inside those calls.
    pub nanos: u64,
    /// `sync_data` calls that returned an error.
    pub failures: u64,
}

/// Read the process-wide WAL fsync counters. Lock-free.
pub fn wal_fsync_stats() -> WalFsyncStats {
    WalFsyncStats {
        count: FSYNC_TOTAL.load(Ordering::Relaxed),
        nanos: FSYNC_NANOS_TOTAL.load(Ordering::Relaxed),
        failures: FSYNC_FAILURES_TOTAL.load(Ordering::Relaxed),
    }
}

/// Error every durability claim carries once the WAL is poisoned. Kept as
/// a constant so tests can assert on the exact contract wording.
const WAL_POISONED_MSG: &str = "WAL poisoned by an earlier fsync failure; commits can no \
     longer be made durable (the OS may have dropped the unsynced pages). Restart the \
     process to recover from the on-disk log";

/// Addresses of the [`WalSyncShared`]s whose next fsync must fail
/// (each entry self-disarms when it fires). There is no portable way to make
/// a real `fdatasync` return EIO/ENOSPC on demand, and the post-failure
/// behavior — poison, never retry — is exactly what the fsyncgate tests have
/// to observe. Scoped per WAL so concurrent tests' fsyncs cannot steal each
/// other's injection (a single shared slot let a second test's arm clobber
/// the first's), yet global (not thread-local) so the Normal-mode background
/// flusher thread can trip it too. [`Wal::drop`] removes its own entry, so a
/// test that arms but never fsyncs cannot leak an arm onto whatever WAL the
/// allocator later places at the same address.
#[cfg(any(test, feature = "testing"))]
static WAL_FSYNC_FAILPOINTS: std::sync::Mutex<Vec<usize>> = std::sync::Mutex::new(Vec::new());

/// Run `sync_data` on `file`, recording its duration in the process-wide
/// counters. Every WAL fsync goes through here so the accounting cannot drift
/// from the actual calls.
fn timed_sync_data(file: &File) -> io::Result<()> {
    let started = std::time::Instant::now();
    let result = file.sync_data();
    match &result {
        Ok(()) => {
            FSYNC_TOTAL.fetch_add(1, Ordering::Relaxed);
            FSYNC_NANOS_TOTAL.fetch_add(started.elapsed().as_nanos() as u64, Ordering::Relaxed);
        }
        Err(_) => {
            FSYNC_FAILURES_TOTAL.fetch_add(1, Ordering::Relaxed);
        }
    }
    result
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum WalRecordType {
    Insert = 1,
    Update = 2,
    Delete = 3,
    Commit = 4,
    Rollback = 5,
    DdlCreateTable = 6,
    DdlDropTable = 7,
    DdlAddColumn = 8,
    DdlDropColumn = 9,
    Begin = 10,
    /// Physical log of one overflow-chain chunk (door D4). Payload:
    /// `page_id u32 | next_page u32 | chunk_len u16 | chunk bytes`. Replayed
    /// by page id under the per-page LSN skip, so it is idempotent.
    OverflowWrite = 11,
    /// Batch of overflow pages returned to the free list (door D4). Payload:
    /// `count u32 | page_id u32 x count`. Idempotent on replay.
    OverflowFree = 12,
}

impl WalRecordType {
    pub fn from_u8(v: u8) -> Option<Self> {
        match v {
            1 => Some(WalRecordType::Insert),
            2 => Some(WalRecordType::Update),
            3 => Some(WalRecordType::Delete),
            4 => Some(WalRecordType::Commit),
            5 => Some(WalRecordType::Rollback),
            6 => Some(WalRecordType::DdlCreateTable),
            7 => Some(WalRecordType::DdlDropTable),
            8 => Some(WalRecordType::DdlAddColumn),
            9 => Some(WalRecordType::DdlDropColumn),
            10 => Some(WalRecordType::Begin),
            11 => Some(WalRecordType::OverflowWrite),
            12 => Some(WalRecordType::OverflowFree),
            _ => None,
        }
    }
}

pub const WAL_MAGIC: &[u8; 4] = b"PWAL";
pub const WAL_FORMAT_VERSION: u16 = 1;
const WAL_FILE_HEADER_SIZE: u64 = 8;

/// WAL record header: len(4) + crc32(4) + tx_id(8) + type(1) + lsn(8) = 25 bytes
const WAL_HEADER_SIZE: usize = 25;

fn write_wal_file_header(file: &mut File) -> io::Result<()> {
    file.seek(SeekFrom::Start(0))?;
    file.write_all(WAL_MAGIC)?;
    file.write_all(&WAL_FORMAT_VERSION.to_le_bytes())?;
    file.write_all(&0u16.to_le_bytes())?;
    file.seek(SeekFrom::End(0))?;
    Ok(())
}

fn wal_records_start(file: &mut File) -> io::Result<u64> {
    let len = file.metadata()?.len();
    if len == 0 {
        write_wal_file_header(file)?;
        return Ok(WAL_FILE_HEADER_SIZE);
    }
    if len >= WAL_FILE_HEADER_SIZE {
        file.seek(SeekFrom::Start(0))?;
        let mut hdr = [0u8; WAL_FILE_HEADER_SIZE as usize];
        file.read_exact(&mut hdr)?;
        if &hdr[0..4] == WAL_MAGIC {
            let version = u16::from_le_bytes(hdr[4..6].try_into().expect("2-byte WAL version"));
            if version != WAL_FORMAT_VERSION {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    format!("unsupported WAL format version: {version}"),
                ));
            }
            return Ok(WAL_FILE_HEADER_SIZE);
        }
    }
    // Legacy 0.4.x WAL: no file header; records start at byte 0.
    Ok(0)
}

/// Maximum allowed size for a single WAL record's data payload.
/// Records claiming more than 256 MB are treated as corruption and
/// stop replay — this prevents a crafted WAL from causing a
/// multi-gigabyte allocation before the CRC check can reject it.
const MAX_WAL_RECORD_SIZE: usize = 256 * 1024 * 1024;

#[derive(Debug)]
pub struct WalRecord {
    pub tx_id: u64,
    pub record_type: WalRecordType,
    /// Monotonic log sequence number assigned at append time. Used by
    /// the page-level idempotent replay: if a page's on-disk LSN is
    /// `>=` this record's LSN, the record has already been applied and
    /// replay skips it.
    pub lsn: u64,
    pub data: Vec<u8>,
}

/// Durability mode for the WAL — analogous to SQLite's `PRAGMA synchronous`
/// combined with `journal_mode=OFF`.
///
/// * `Full` — every mutation appends a record and `flush()` calls
///   `sync_data()` so the OS guarantees the bytes hit stable storage before
///   the call returns. This is the default and the only safe choice when
///   crash recovery must be perfect.
///
/// * `Off`  — every `append()` and `flush()` is a zero-work no-op. No CRC,
///   no BufWriter, no fsync, no recovery. This matches SQLite's `:memory:`
///   semantics and is the only way to compare apples-to-apples against
///   in-memory engines in benches. Never use this in production — a crash
///   loses every mutation since the last `Catalog::checkpoint()`.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
pub enum WalSyncMode {
    #[default]
    Full,
    /// `Normal` — every commit buffers its record through to the OS
    /// (`BufWriter::flush`, so the bytes are file-visible) and returns
    /// WITHOUT an fsync; a background flusher fsyncs on a fixed interval
    /// (`NORMAL_FSYNC_INTERVAL`). A *process* crash loses nothing (replay
    /// reads the bytes already in the OS page cache); an *OS* crash / power
    /// loss can lose only the unsynced tail (≤ one interval of writes). This
    /// is SQLite `synchronous=NORMAL` / Postgres `synchronous_commit=off`
    /// semantics: opt-in, bounded-loss, and ~15–40× faster single-row writes
    /// because the fsync leaves the commit/lock path.
    Normal,
    Off,
}

/// How often the background flusher fsyncs in [`WalSyncMode::Normal`]. This is
/// the upper bound on the crash-loss window (OS-crash / power-loss only).
const NORMAL_FSYNC_INTERVAL: Duration = Duration::from_millis(10);

/// Fsync-coordination state shared between the `Wal`, the Normal-mode
/// background flusher, and any outstanding [`WalDurabilityTicket`]s.
///
/// This is the heart of Full-mode group commit: `dirty_gen` counts
/// flush-to-OS generations, `synced_gen` tracks the highest generation an
/// fsync has covered, and `sync_file` is both the fd fsyncs go through and
/// the leader election — whoever holds the mutex fsyncs on behalf of every
/// generation registered before the fsync started.
#[derive(Debug)]
struct WalSyncShared {
    /// Monotonic counter bumped on every durable-intent flush-to-OS (non-Off).
    /// A generation is only registered after its bytes reached the OS file
    /// (`BufWriter::flush`), so an fsync issued afterwards always covers it.
    dirty_gen: AtomicU64,
    /// Highest `dirty_gen` value known to be fsync-durable. Advanced by
    /// group-commit leaders and by the Normal background flusher.
    synced_gen: AtomicU64,
    /// Number of `sync_data` calls issued on the WAL file. Test/metrics hook:
    /// group-commit coalescing shows up as fewer fsyncs than commits.
    fsync_count: AtomicU64,
    /// The fd used for fsyncs, doubling as the group-commit leader lock.
    /// `None` only if cloning the writer's fd failed on (re)open.
    sync_file: Mutex<Option<File>>,
    /// Sticky fsyncgate latch. Once an fsync on this WAL has failed, the OS
    /// may have dropped the dirty pages and marked them clean, so a RETRIED
    /// fsync would report success without the bytes ever reaching stable
    /// storage. Set on the first failure and never cleared: every later
    /// durability claim fails fast with [`WAL_POISONED_MSG`] instead of
    /// issuing another fsync. Recovery is a process restart (WAL replay
    /// reads the log that actually made it to disk).
    sync_poisoned: std::sync::atomic::AtomicBool,
}

impl WalSyncShared {
    /// True (once) when the fsync failpoint is armed for THIS shared state.
    #[cfg(any(test, feature = "testing"))]
    fn take_fsync_failpoint(&self) -> bool {
        let me = self as *const WalSyncShared as usize;
        let mut armed = WAL_FSYNC_FAILPOINTS.lock().unwrap();
        match armed.iter().position(|&addr| addr == me) {
            Some(i) => {
                armed.remove(i);
                true
            }
            None => false,
        }
    }

    /// The one fsync choke point for this WAL: [`timed_sync_data`] plus the
    /// test-only failure injection.
    fn fsync_file(&self, file: &File) -> io::Result<()> {
        #[cfg(any(test, feature = "testing"))]
        if self.take_fsync_failpoint() {
            FSYNC_FAILURES_TOTAL.fetch_add(1, Ordering::Relaxed);
            return Err(io::Error::other("injected WAL fsync failure"));
        }
        timed_sync_data(file)
    }

    fn new(sync_file: Option<File>) -> Self {
        WalSyncShared {
            dirty_gen: AtomicU64::new(0),
            synced_gen: AtomicU64::new(0),
            fsync_count: AtomicU64::new(0),
            sync_file: Mutex::new(sync_file),
            sync_poisoned: std::sync::atomic::AtomicBool::new(false),
        }
    }

    /// Block until an fsync covering `gen` has completed (leader/follower
    /// group commit). The first caller to take the lock fsyncs once for every
    /// generation registered so far; callers queued behind it wake up already
    /// covered and return without an fsync of their own. A lone caller finds
    /// the lock free and fsyncs immediately — group commit never introduces a
    /// wait for company.
    fn sync_until(&self, gen: u64) -> io::Result<()> {
        if self.synced_gen.load(Ordering::Acquire) >= gen {
            return Ok(());
        }
        let guard = self
            .sync_file
            .lock()
            .map_err(|_| io::Error::other("WAL sync lock poisoned"))?;
        // A leader that ran while we were queued may already have covered us.
        if self.synced_gen.load(Ordering::Acquire) >= gen {
            return Ok(());
        }
        // Checked under the leader lock, after the covered fast paths: a
        // generation an earlier fsync already covered IS durable and may
        // truthfully return Ok even on a poisoned WAL.
        if self.sync_poisoned.load(Ordering::Acquire) {
            return Err(io::Error::other(WAL_POISONED_MSG));
        }
        let file = guard
            .as_ref()
            .ok_or_else(|| io::Error::other("WAL sync fd unavailable"))?;
        // Snapshot BEFORE the fsync: every generation registered by now has
        // its bytes in the OS file already, so this one fsync covers them all.
        let cover = self.dirty_gen.load(Ordering::Acquire);
        if let Err(e) = self.fsync_file(file) {
            // fsyncgate: retrying after a failed fsync is unsound (the OS
            // may have dropped the pages and would report the retry as
            // success). Poison the WAL permanently and surface the failure.
            self.sync_poisoned.store(true, Ordering::Release);
            tracing::error!(
                error = %e,
                "WAL fsync failed; poisoning the WAL — no further commits \
                 will be acknowledged until the process restarts"
            );
            return Err(e);
        }
        self.fsync_count.fetch_add(1, Ordering::Relaxed);
        self.synced_gen.fetch_max(cover, Ordering::AcqRel);
        Ok(())
    }

    /// Swap the fsync fd and mark every generation registered so far as
    /// settled. Called when the WAL file is truncated or recreated: the bytes
    /// those generations covered are gone from the log — either already
    /// durable elsewhere (checkpoint flushed the heaps; the discard paths
    /// `sync_data` the truncated file) or intentionally discarded by rollback
    /// — so no ticket must ever block on them again.
    fn replace_file(&self, file: Option<File>) {
        // Take the leader lock so an in-flight fsync on the old fd finishes
        // before the swap. Poisoning is impossible in practice (the critical
        // section cannot panic) but recover anyway rather than propagate.
        let mut guard = match self.sync_file.lock() {
            Ok(g) => g,
            Err(poisoned) => poisoned.into_inner(),
        };
        let d = self.dirty_gen.load(Ordering::Acquire);
        self.synced_gen.fetch_max(d, Ordering::AcqRel);
        *guard = file;
    }
}

/// A claim on WAL durability handed out by a deferred Full-mode flush: the
/// commit's records have reached the OS file but are not yet guaranteed on
/// stable storage. [`Self::wait`] blocks until an fsync covering them has
/// completed — the caller must not acknowledge the commit before `wait`
/// returns `Ok(())`.
///
/// Tickets are cumulative: generations are registered in order, so waiting on
/// a later ticket also makes every earlier generation durable. Waiting takes
/// no `Wal` lock, which is what lets a committer release the engine's write
/// lock first and other committers append while the fsync runs — the overlap
/// that lets one fsync cover many commits.
#[derive(Debug)]
#[must_use = "a commit must not be acknowledged until wait() returns Ok"]
pub struct WalDurabilityTicket {
    gen: u64,
    shared: Arc<WalSyncShared>,
}

impl WalDurabilityTicket {
    /// Block until an fsync covering this ticket's WAL records has completed.
    /// See [`WalSyncShared::sync_until`] for the leader/follower scheme.
    pub fn wait(self) -> io::Result<()> {
        self.shared.sync_until(self.gen)
    }
}

pub struct Wal {
    path: PathBuf,
    writer: Option<BufWriter<File>>,
    batch_size: usize,
    pending: usize,
    sync_mode: WalSyncMode,
    /// Monotonic LSN counter. Starts at 1 (0 means "no WAL record has
    /// ever touched this page") and increments by 1 on every `append`.
    next_lsn: u64,
    /// File length as of the last successful WAL sync/truncate/open.
    ///
    /// `BufWriter` may write large pending records through to the OS file
    /// before [`Self::flush`] is called. Those bytes are file-visible but
    /// not transaction-durable. Rollback truncates back to this boundary so
    /// a same-process reopen cannot replay uncommitted records.
    records_start: u64,
    synced_len: u64,
    /// Group-commit fsync coordination (see [`WalSyncShared`]).
    shared: Arc<WalSyncShared>,
    /// When `true`, a Full-mode `flush()` registers the generation it needs
    /// durable instead of fsyncing inline; [`Self::take_durability_ticket`]
    /// hands the claim to the caller, who must wait on it before
    /// acknowledging the commit. See [`Self::set_defer_sync`].
    defer_sync: bool,
    /// Highest generation registered by deferred flushes since the last
    /// `take_durability_ticket`. Cumulative — a later generation covers all
    /// earlier ones, so overwriting never loses coverage.
    deferred_gen: Option<u64>,
    /// Background fsync thread; present only while in `Normal` mode.
    flusher: Option<Flusher>,
}

/// Background fsync worker for [`WalSyncMode::Normal`]. Owns a cloned WAL file
/// descriptor and fsyncs it on [`NORMAL_FSYNC_INTERVAL`] whenever new bytes
/// have been buffered, keeping the fsync off the commit/lock path. fsync on the
/// cloned fd flushes the same underlying file (inode) the writer appends to.
struct Flusher {
    handle: Option<JoinHandle<()>>,
    /// `(stop, condvar)` — set `stop=true` + notify to wake the thread early.
    ctl: Arc<(Mutex<bool>, Condvar)>,
}

impl Flusher {
    fn spawn(file: File, shared: Arc<WalSyncShared>, interval: Duration) -> Flusher {
        let ctl: Arc<(Mutex<bool>, Condvar)> = Arc::new((Mutex::new(false), Condvar::new()));
        let ctl_thread = Arc::clone(&ctl);
        let handle = std::thread::Builder::new()
            .name("powdb-wal-flusher".into())
            .spawn(move || {
                let (lock, cvar) = &*ctl_thread;
                loop {
                    let stopping = {
                        let stop = lock.lock().expect("wal flusher lock");
                        if *stop {
                            true
                        } else {
                            let (stop, _timeout) =
                                cvar.wait_timeout(stop, interval).expect("wal flusher wait");
                            *stop
                        }
                    };
                    // A concurrent Full-mode leader may have poisoned the
                    // WAL between ticks; a poisoned WAL has nothing left for
                    // a flusher to do.
                    if shared.sync_poisoned.load(Ordering::Acquire) {
                        break;
                    }
                    // fsync if the writer has buffered new bytes since last sync.
                    let d = shared.dirty_gen.load(Ordering::Acquire);
                    if d > shared.synced_gen.load(Ordering::Acquire) {
                        match shared.fsync_file(&file) {
                            Ok(()) => {
                                shared.fsync_count.fetch_add(1, Ordering::Relaxed);
                                // fetch_max, not store: a Full-mode group
                                // commit may have advanced past `d` between
                                // the load and the fsync (mode switches).
                                shared.synced_gen.fetch_max(d, Ordering::AcqRel);
                            }
                            // In Normal mode this background fsync is the ONLY
                            // durability point, and retrying a failed fsync is
                            // unsound (fsyncgate: the OS may have dropped the
                            // pages and would report the retry as success).
                            // Poison the WAL — later flushes fail fast with
                            // WAL_POISONED_MSG — and stop flushing.
                            Err(e) => {
                                shared.sync_poisoned.store(true, Ordering::Release);
                                tracing::error!(
                                    error = %e,
                                    "WAL background fsync failed; poisoning the WAL and \
                                     stopping the flusher — no further commits will be \
                                     acknowledged until the process restarts"
                                );
                                break;
                            }
                        }
                    }
                    if stopping {
                        break;
                    }
                }
            })
            .expect("spawn wal flusher thread");
        Flusher {
            handle: Some(handle),
            ctl,
        }
    }

    fn stop(&mut self) {
        {
            let (lock, cvar) = &*self.ctl;
            let mut stop = lock.lock().expect("wal flusher lock");
            *stop = true;
            cvar.notify_all();
        }
        if let Some(h) = self.handle.take() {
            let _ = h.join();
        }
    }
}

impl Drop for Flusher {
    fn drop(&mut self) {
        self.stop();
    }
}

impl Wal {
    pub fn create(path: &Path, batch_size: usize) -> io::Result<Self> {
        let mut file = OpenOptions::new()
            .create(true)
            .write(true)
            .read(true)
            .truncate(true)
            .open(path)?;
        write_wal_file_header(&mut file)?;
        let sync_fd = file.try_clone()?;
        Ok(Wal {
            path: path.to_path_buf(),
            writer: Some(BufWriter::new(file)),
            batch_size,
            pending: 0,
            sync_mode: WalSyncMode::default(),
            next_lsn: 1,
            records_start: WAL_FILE_HEADER_SIZE,
            synced_len: WAL_FILE_HEADER_SIZE,
            shared: Arc::new(WalSyncShared::new(Some(sync_fd))),
            defer_sync: false,
            deferred_gen: None,
            flusher: None,
        })
    }

    pub fn open(path: &Path, batch_size: usize) -> io::Result<Self> {
        let mut file = OpenOptions::new()
            .create(true)
            .read(true)
            .append(true)
            .open(path)?;
        let records_start = wal_records_start(&mut file)?;
        let synced_len = file.metadata()?.len();
        let sync_fd = file.try_clone()?;
        Ok(Wal {
            path: path.to_path_buf(),
            writer: Some(BufWriter::new(file)),
            batch_size,
            pending: 0,
            sync_mode: WalSyncMode::default(),
            next_lsn: 1,
            records_start,
            synced_len,
            shared: Arc::new(WalSyncShared::new(Some(sync_fd))),
            defer_sync: false,
            deferred_gen: None,
            flusher: None,
        })
    }

    /// Toggle the durability mode. See [`WalSyncMode`] for the contract.
    /// Starts the background flusher when entering `Normal`, and stops it when
    /// leaving `Normal`. The fsync-behavior change takes effect on the next
    /// `flush()`.
    pub fn set_sync_mode(&mut self, mode: WalSyncMode) {
        if mode == self.sync_mode {
            return;
        }
        self.sync_mode = mode;
        match mode {
            WalSyncMode::Normal => self.start_flusher(),
            WalSyncMode::Full | WalSyncMode::Off => self.stop_flusher(),
        }
    }

    /// Spawn the Normal-mode background flusher if not already running. The
    /// flusher fsyncs a cloned WAL fd, so it never contends on the writer.
    fn start_flusher(&mut self) {
        if self.flusher.is_some() {
            return;
        }
        if let Some(writer) = self.writer.as_ref() {
            if let Ok(file) = writer.get_ref().try_clone() {
                self.flusher = Some(Flusher::spawn(
                    file,
                    Arc::clone(&self.shared),
                    NORMAL_FSYNC_INTERVAL,
                ));
            }
        }
    }

    /// Stop the background flusher (final fsync + join), if running.
    fn stop_flusher(&mut self) {
        if let Some(mut f) = self.flusher.take() {
            f.stop();
        }
    }

    /// Arm the fsync failpoint for this WAL: its next fsync — from any
    /// thread, including the Normal-mode background flusher — fails with an
    /// injected error. Self-disarming.
    #[cfg(any(test, feature = "testing"))]
    pub(crate) fn arm_fsync_failpoint(&self) {
        let me = Arc::as_ptr(&self.shared) as usize;
        let mut armed = WAL_FSYNC_FAILPOINTS.lock().unwrap();
        if !armed.contains(&me) {
            armed.push(me);
        }
    }

    /// Remove this WAL's armed failpoint, if any (see the static's docs).
    #[cfg(any(test, feature = "testing"))]
    fn disarm_fsync_failpoint(&self) {
        let me = Arc::as_ptr(&self.shared) as usize;
        WAL_FSYNC_FAILPOINTS.lock().unwrap().retain(|&a| a != me);
    }

    /// True once any fsync on this WAL has failed. The WAL is then
    /// permanently poisoned — every later durability claim fails fast with
    /// the same error instead of retrying the fsync (fsyncgate: a retried
    /// fsync can report success over pages the OS already dropped). Restart
    /// the process to recover from the on-disk log.
    pub fn is_sync_poisoned(&self) -> bool {
        self.shared
            .sync_poisoned
            .load(std::sync::atomic::Ordering::Acquire)
    }

    /// The highest dirty generation known to be fsync-durable. Advances on
    /// every Full commit and on each Normal background-flusher cycle. Exposed
    /// for tests and (future) metrics.
    pub fn synced_generation(&self) -> u64 {
        self.shared.synced_gen.load(Ordering::Acquire)
    }

    /// Number of fsyncs issued against the WAL file (group-commit leaders,
    /// inline Full-mode flushes, and the Normal background flusher). Exposed
    /// for tests and (future) metrics: group-commit coalescing shows up as
    /// fewer fsyncs than commits.
    pub fn fsync_count(&self) -> u64 {
        self.shared.fsync_count.load(Ordering::Relaxed)
    }

    /// Defer Full-mode commit fsyncs. While enabled, [`Self::flush`]
    /// registers the generation it needs durable instead of fsyncing inline;
    /// the pending claim is retrieved with [`Self::take_durability_ticket`]
    /// and the caller must wait on it before acknowledging the commit. This
    /// is how group commit lets the fsync leave the engine's exclusive-lock
    /// hold: append + register under the lock, wait after releasing it.
    ///
    /// `Normal` and `Off` modes are unaffected (they never fsync inline).
    pub fn set_defer_sync(&mut self, defer: bool) {
        self.defer_sync = defer;
    }

    /// Whether Full-mode commit fsyncs are currently deferred.
    pub fn defer_sync(&self) -> bool {
        self.defer_sync
    }

    /// Take the durability claim registered by deferred flushes since the
    /// last take, if any. Generations are cumulative, so one ticket covers
    /// every deferred flush that happened before it was taken.
    pub fn take_durability_ticket(&mut self) -> Option<WalDurabilityTicket> {
        self.deferred_gen.take().map(|gen| WalDurabilityTicket {
            gen,
            shared: Arc::clone(&self.shared),
        })
    }

    /// Returns the current sync mode (used by tests + introspection).
    pub fn sync_mode(&self) -> WalSyncMode {
        self.sync_mode
    }

    /// `true` when the WAL is in [`WalSyncMode::Off`] — i.e. every
    /// `append`/`flush` is a no-op. Catalog mutation hot paths check
    /// this BEFORE constructing WAL payloads so we don't pay
    /// `encode_row_into` + `encode_wal_payload` allocs only to throw
    /// the result away inside `append`. This is the difference between
    /// "no fsync" and "free" — the former is still 50–60% slower than
    /// the no-WAL baseline on `update_by_filter`/`delete_by_filter`,
    /// the latter matches the baseline.
    #[inline]
    pub fn is_off(&self) -> bool {
        matches!(self.sync_mode, WalSyncMode::Off)
    }

    /// LSN of the most recently appended record, or 0 if nothing has
    /// been appended yet (or the WAL is off).
    ///
    /// Used by schema-change paths to capture a "barrier LSN" that
    /// reflects the DDL record's position in the log; the heap can then
    /// stamp its pages with that LSN so replay skips every
    /// Insert/Update/Delete that pre-dates the schema change (those rows
    /// have already been migrated to the new layout in place).
    #[inline]
    pub fn last_appended_lsn(&self) -> u64 {
        if matches!(self.sync_mode, WalSyncMode::Off) {
            return 0;
        }
        self.next_lsn.saturating_sub(1)
    }

    /// Ensure the next LSN this WAL hands out is at least `lsn`. Called on
    /// open, after recovery, to restore monotonicity: heap pages carry the
    /// LSNs stamped during replay (and by DDL rewrites), but `Wal::open`
    /// always resets `next_lsn` to 1. Without this, writes taken after a
    /// crash-recovery would reuse LSNs at or below those stamped page LSNs,
    /// and the next crash's replay would skip them as already-applied —
    /// silent data loss. Never lowers the counter.
    pub fn set_next_lsn_at_least(&mut self, lsn: u64) {
        if lsn > self.next_lsn {
            self.next_lsn = lsn;
        }
    }

    /// Append a record to the WAL buffer. Auto-flushes when batch is full.
    ///
    /// In [`WalSyncMode::Off`] this is a zero-work no-op — see the enum's
    /// doc for the durability contract.
    pub fn append(
        &mut self,
        tx_id: u64,
        record_type: WalRecordType,
        data: &[u8],
    ) -> io::Result<()> {
        if matches!(self.sync_mode, WalSyncMode::Off) {
            return Ok(());
        }
        let lsn = self.next_lsn;
        self.next_lsn += 1;
        let total_len = (WAL_HEADER_SIZE + data.len()) as u32;

        // Compute CRC over tx_id + type + lsn + data
        let mut crc_input = Vec::with_capacity(17 + data.len());
        crc_input.extend_from_slice(&tx_id.to_le_bytes());
        crc_input.push(record_type as u8);
        crc_input.extend_from_slice(&lsn.to_le_bytes());
        crc_input.extend_from_slice(data);
        let crc = crc32fast::hash(&crc_input);

        // Write: len + crc + tx_id + type + lsn + data
        let writer = self
            .writer
            .as_mut()
            .ok_or_else(|| io::Error::other("WAL writer unavailable"))?;
        writer.write_all(&total_len.to_le_bytes())?;
        writer.write_all(&crc.to_le_bytes())?;
        writer.write_all(&tx_id.to_le_bytes())?;
        writer.write_all(&[record_type as u8])?;
        writer.write_all(&lsn.to_le_bytes())?;
        writer.write_all(data)?;

        self.pending += 1;
        if self.pending >= self.batch_size {
            self.flush()?;
        }
        Ok(())
    }

    /// Flush buffered records to disk (the group commit point).
    ///
    /// In `Full` mode the commit is durable when this returns: the buffered
    /// bytes are pushed to the OS and an fsync covering them completes before
    /// the call returns — unless durability deferral is active (see
    /// [`Self::set_defer_sync`]), in which case the fsync obligation is
    /// registered and handed to the caller via
    /// [`Self::take_durability_ticket`]. Either way, concurrent committers'
    /// fsyncs coalesce: one fsync covers every generation registered before
    /// it started, and a lone committer fsyncs immediately (no batching
    /// delay).
    ///
    /// No-op if nothing has been appended since the last flush. This makes
    /// it safe for the executor to unconditionally call `sync_wal` at the
    /// end of every statement — read queries pay zero fsync cost.
    pub fn flush(&mut self) -> io::Result<()> {
        let Some(gen) = self.flush_to_os()? else {
            return Ok(());
        };
        // SQLite-style synchronous knob: only the fsync is gated on the mode.
        // The flush-to-OS above always runs so a process crash still recovers
        // cleanly via `read_all`. In `Full` the fsync happens here (or via
        // the deferred ticket); in `Normal` the background flusher fsyncs off
        // this path.
        if matches!(self.sync_mode, WalSyncMode::Full) {
            if self.defer_sync {
                // Cumulative: the newest generation covers all earlier ones,
                // so overwriting an untaken claim never loses coverage.
                self.deferred_gen = Some(gen);
            } else {
                self.shared.sync_until(gen)?;
            }
        }
        Ok(())
    }

    /// Settle all written WAL generations before committed heap pages can be
    /// persisted, including claims already handed to deferred commit callers.
    pub(crate) fn sync_before_heap_flush(&mut self) -> io::Result<()> {
        self.flush_to_os()?;
        if !matches!(self.sync_mode, WalSyncMode::Off) {
            self.shared
                .sync_until(self.shared.dirty_gen.load(Ordering::Acquire))?;
        }
        Ok(())
    }

    /// Push buffered records through to the OS file (no fsync) and register
    /// the resulting dirty generation. Returns `Ok(None)` when there was
    /// nothing pending or the WAL is `Off`.
    fn flush_to_os(&mut self) -> io::Result<Option<u64>> {
        let batch = self.pending;
        if batch == 0 {
            return Ok(None);
        }
        // Poisoned WALs take no new durable-intent bytes: in Normal mode the
        // ack happens right after this flush-to-OS, and the mechanism that
        // would have made it durable is permanently gone.
        if !matches!(self.sync_mode, WalSyncMode::Off)
            && self.shared.sync_poisoned.load(Ordering::Acquire)
        {
            return Err(io::Error::other(WAL_POISONED_MSG));
        }
        // Borrow the writer only for the I/O, then drop it before touching the
        // generation counters (which borrow `self`).
        let new_len = {
            let writer = self
                .writer
                .as_mut()
                .ok_or_else(|| io::Error::other("WAL writer unavailable"))?;
            writer.flush()?;
            writer.get_ref().metadata()?.len()
        };
        self.synced_len = new_len;
        self.pending = 0;
        if matches!(self.sync_mode, WalSyncMode::Off) {
            return Ok(None);
        }
        // Registered only after the bytes are OS-visible, so any fsync issued
        // from here on covers this generation.
        let gen = self.shared.dirty_gen.fetch_add(1, Ordering::Release) + 1;
        debug!(records = batch, "wal group commit");
        Ok(Some(gen))
    }

    /// True when records have been appended to the in-memory WAL buffer
    /// since the last durable flush.
    #[inline]
    pub fn has_pending(&self) -> bool {
        self.pending > 0
    }

    /// Flush pending WAL bytes, then return the durable file length. Used as
    /// an explicit-transaction rollback boundary.
    pub fn synced_len(&mut self) -> io::Result<u64> {
        self.flush()?;
        Ok(self.synced_len)
    }

    /// Discard buffered (not-yet-flushed) WAL bytes and truncate the durable
    /// log back to `len`. This is intentionally not implemented by dropping
    /// the existing BufWriter: BufWriter's Drop attempts to flush buffered
    /// bytes, which would resurrect rolled-back records.
    pub fn discard_and_truncate_to(&mut self, len: u64) -> io::Result<()> {
        if matches!(self.sync_mode, WalSyncMode::Off) {
            self.pending = 0;
            self.synced_len = len;
            return Ok(());
        }

        if let Some(writer) = self.writer.take() {
            let (_file, _buffer_result) = writer.into_parts();
        }

        let mut file = OpenOptions::new()
            .create(true)
            .read(true)
            .append(true)
            .open(&self.path)?;
        file.set_len(len)?;
        file.seek(SeekFrom::End(0))?;
        file.sync_data()?;
        let sync_fd = file.try_clone()?;
        self.writer = Some(BufWriter::new(file));
        // Everything that survived the truncation was just `sync_data`ed
        // above, and everything past `len` is intentionally discarded; settle
        // all registered generations and drop any deferred claim.
        self.deferred_gen = None;
        self.shared.replace_file(Some(sync_fd));
        self.pending = 0;
        self.synced_len = len;
        Ok(())
    }

    /// Read all valid records from the WAL file.
    pub fn read_all(&self) -> io::Result<Vec<WalRecord>> {
        self.read_through_len(u64::MAX)
    }

    /// Read valid records up to a byte length boundary in the WAL file.
    pub fn read_through_len(&self, max_len: u64) -> io::Result<Vec<WalRecord>> {
        parse_wal_records(&self.path, max_len)
    }
}

/// Parse valid WAL records from `path` up to `max_len` bytes, opening the file
/// **read-only**. This is the single record-decoding loop shared by
/// [`Wal::read_through_len`] and the read-only open path
/// ([`read_records_at_path`]); neither ever writes to the file, so it is safe on
/// a directory opened for read-only snapshot serving.
fn parse_wal_records(path: &Path, max_len: u64) -> io::Result<Vec<WalRecord>> {
    let mut file = File::open(path)?;
    let file_len = file.metadata()?.len().min(max_len);
    let mut file_for_header = File::open(path)?;
    let mut pos = wal_records_start_readonly(&mut file_for_header)?;
    let mut records = Vec::new();

    loop {
        match parse_record_at(&mut file, pos, file_len)? {
            ParseOutcome::Record(record, next_pos) => {
                records.push(record);
                pos = next_pos;
            }
            ParseOutcome::Eof => break,
            ParseOutcome::Corrupt { reason, resume_at } => {
                // A record that will not parse is either the torn tail of the
                // last write (everything after it is nothing) or damage in the
                // middle of the log (records after it survived). Truncating on
                // the second one throws away committed work with no warning,
                // so the two have to be told apart before anything is
                // discarded.
                if let Some(found_at) = probe_for_valid_record(&mut file, resume_at, pos, file_len)?
                {
                    return Err(StorageError::WalReplay(format!(
                        "{}: record at byte {pos} is corrupt ({reason}), but a valid record follows at byte {found_at}: this is damage inside the log, not a torn tail",
                        path.display()
                    ))
                    .into());
                }
                tracing::warn!(
                    path = %path.display(),
                    offset = pos,
                    discarded_bytes = file_len.saturating_sub(pos),
                    reason,
                    "discarding torn WAL tail"
                );
                break;
            }
        }
    }

    Ok(records)
}

/// How far past a corrupt record to look for the next valid one when the
/// corrupt record's own length field cannot be trusted. One megabyte covers
/// any plausible run of damage while keeping the scan bounded on a
/// multi-hundred-megabyte log.
const CORRUPTION_PROBE_WINDOW: u64 = 1024 * 1024;

/// Look for a valid record after the corrupt one at `bad_pos`. Returns the
/// offset of the first valid record found, or `None` when nothing after the
/// damage parses (a torn tail).
///
/// The first probe is the cheap and overwhelmingly common one: bit rot inside
/// a record leaves its length field intact, so the next record still starts
/// exactly where that length says. Only when that fails does this fall back to
/// a bounded byte scan, which is what catches a corrupt length field.
fn probe_for_valid_record(
    file: &mut File,
    resume_at: Option<u64>,
    bad_pos: u64,
    file_len: u64,
) -> io::Result<Option<u64>> {
    if let Some(next) = resume_at {
        if next <= file_len
            && matches!(
                parse_record_at(file, next, file_len)?,
                ParseOutcome::Record(..)
            )
        {
            return Ok(Some(next));
        }
    }
    let scan_end = file_len
        .saturating_sub(WAL_HEADER_SIZE as u64)
        .min(bad_pos.saturating_add(CORRUPTION_PROBE_WINDOW));
    let mut candidate = bad_pos.saturating_add(1);
    while candidate <= scan_end {
        if matches!(
            parse_record_at(file, candidate, file_len)?,
            ParseOutcome::Record(..)
        ) {
            return Ok(Some(candidate));
        }
        candidate += 1;
    }
    Ok(None)
}

/// Whether `path` holds a WAL with no `PWAL` file header.
///
/// Such a file is read as a pre-v0.5.0 headerless log whose records start at
/// byte 0. That is the right reading for a genuinely old directory and the
/// wrong one everywhere else: it parses arbitrary bytes as records and, when
/// the first of them does not decode, silently drops every un-checkpointed
/// row. Callers that know the directory is newer than that refuse instead.
pub fn is_headerless_log(path: &Path) -> io::Result<bool> {
    let mut file = match File::open(path) {
        Ok(file) => file,
        Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
        Err(error) => return Err(error),
    };
    let len = file.metadata()?.len();
    if len == 0 {
        // A fresh or freshly truncated WAL: `Wal::open` writes the header.
        return Ok(false);
    }
    if len < WAL_FILE_HEADER_SIZE {
        return Ok(true);
    }
    let mut magic = [0u8; 4];
    file.read_exact(&mut magic)?;
    Ok(&magic != WAL_MAGIC)
}

/// Why a record could not be parsed, and where the log claims the next record
/// begins.
enum ParseOutcome {
    Record(WalRecord, u64),
    Eof,
    Corrupt {
        reason: &'static str,
        /// The offset the corrupt record's own length field points at, when it
        /// is inside the file. `None` when the length itself is unusable.
        resume_at: Option<u64>,
    },
}

/// Parse the single record starting at `pos`.
///
/// Returns [`ParseOutcome::Eof`] when there is nothing left to read and
/// [`ParseOutcome::Corrupt`] when a record is there but will not decode. The
/// caller decides whether that is a torn tail or damage inside the log.
fn parse_record_at(file: &mut File, pos: u64, file_len: u64) -> io::Result<ParseOutcome> {
    if pos + WAL_HEADER_SIZE as u64 > file_len {
        return Ok(ParseOutcome::Eof);
    }
    file.seek(SeekFrom::Start(pos))?;

    let mut header = [0u8; WAL_HEADER_SIZE];
    if file.read_exact(&mut header).is_err() {
        return Ok(ParseOutcome::Eof);
    }

    // These slice-to-array conversions are infallible (fixed-size sub-slices
    // of a 25-byte array) but we avoid `unwrap` to satisfy the project-wide
    // zero-panic policy.
    let corrupt = |reason: &'static str| ParseOutcome::Corrupt {
        reason,
        resume_at: None,
    };
    let total_len_bytes: [u8; 4] = match header[0..4].try_into() {
        Ok(b) => b,
        Err(_) => return Ok(corrupt("unreadable length field")),
    };
    let total_len = u32::from_le_bytes(total_len_bytes) as usize;
    let stored_crc_bytes: [u8; 4] = match header[4..8].try_into() {
        Ok(b) => b,
        Err(_) => return Ok(corrupt("unreadable CRC field")),
    };
    let stored_crc = u32::from_le_bytes(stored_crc_bytes);
    let tx_id_bytes: [u8; 8] = match header[8..16].try_into() {
        Ok(b) => b,
        Err(_) => return Ok(corrupt("unreadable transaction id")),
    };
    let tx_id = u64::from_le_bytes(tx_id_bytes);
    let record_type = match WalRecordType::from_u8(header[16]) {
        Some(rt) => rt,
        None => return Ok(corrupt("unknown record type")),
    };
    let lsn_bytes: [u8; 8] = match header[17..25].try_into() {
        Ok(b) => b,
        Err(_) => return Ok(corrupt("unreadable LSN")),
    };
    let lsn = u64::from_le_bytes(lsn_bytes);

    // TASK-11: Verify the record fits within the file before allocating.
    // Catches truncated writes without any allocation.
    if pos + total_len as u64 > file_len {
        return Ok(corrupt("record extends past the end of the file"));
    }

    // TASK-09: Use checked_sub to prevent integer underflow when a corrupted
    // WAL has total_len < WAL_HEADER_SIZE.
    let data_len = match total_len.checked_sub(WAL_HEADER_SIZE) {
        Some(len) => len,
        None => return Ok(corrupt("length field shorter than a record header")),
    };

    // TASK-10: Cap allocation size before reading data. A crafted WAL claiming
    // a huge total_len would otherwise allocate gigabytes before the CRC check
    // rejects the record.
    if data_len > MAX_WAL_RECORD_SIZE {
        return Ok(corrupt("record payload exceeds the maximum record size"));
    }

    // From here on the length field is usable, so the caller can look for the
    // next record exactly where this one says it ends.
    let next_pos = pos + total_len as u64;

    let mut data = vec![0u8; data_len];
    if data_len > 0 {
        file.read_exact(&mut data)?;
    }

    // Verify CRC (includes lsn in the hash input)
    let mut crc_input = Vec::with_capacity(17 + data.len());
    crc_input.extend_from_slice(&tx_id.to_le_bytes());
    crc_input.push(record_type as u8);
    crc_input.extend_from_slice(&lsn.to_le_bytes());
    crc_input.extend_from_slice(&data);
    if crc32fast::hash(&crc_input) != stored_crc {
        return Ok(ParseOutcome::Corrupt {
            reason: "CRC32 mismatch",
            resume_at: Some(next_pos),
        });
    }

    Ok(ParseOutcome::Record(
        WalRecord {
            tx_id,
            record_type,
            lsn,
            data,
        },
        next_pos,
    ))
}

/// Report whether `path` holds at least one committed WAL record, without
/// ever opening a writable handle. Returns `false` if the file does not
/// exist. Used by the read-only catalog open path to decide whether a
/// directory is quiescent before serving it, without creating, appending
/// to, or truncating the WAL the way [`Wal::open`] would. Only the first
/// record's header and payload are read; a valid first record is proof
/// enough that the directory needs recovery.
pub fn wal_has_committed_records(path: &Path) -> io::Result<bool> {
    if !path.exists() {
        return Ok(false);
    }
    let mut file = File::open(path)?;
    let file_len = file.metadata()?.len();
    let mut file_for_header = File::open(path)?;
    let pos = wal_records_start_readonly(&mut file_for_header)?;
    Ok(matches!(
        parse_record_at(&mut file, pos, file_len)?,
        ParseOutcome::Record(..)
    ))
}

/// Locate the first record byte in a WAL file **without mutating it**. Mirrors
/// [`wal_records_start`] but never writes a header for an empty/headerless file
/// (the read-only path must not touch the directory): it returns the default
/// record start instead.
fn wal_records_start_readonly(file: &mut File) -> io::Result<u64> {
    let len = file.metadata()?.len();
    if len >= WAL_FILE_HEADER_SIZE {
        file.seek(SeekFrom::Start(0))?;
        let mut hdr = [0u8; WAL_FILE_HEADER_SIZE as usize];
        file.read_exact(&mut hdr)?;
        if &hdr[0..4] == WAL_MAGIC {
            let version = u16::from_le_bytes(hdr[4..6].try_into().expect("2-byte WAL version"));
            if version != WAL_FORMAT_VERSION {
                return Err(io::Error::new(
                    io::ErrorKind::InvalidData,
                    format!("unsupported WAL format version: {version}"),
                ));
            }
            return Ok(WAL_FILE_HEADER_SIZE);
        }
        // Legacy 0.4.x WAL: no file header; records start at byte 0.
        return Ok(0);
    }
    // Empty or sub-header file: nothing to read, and we must not write a header.
    Ok(WAL_FILE_HEADER_SIZE.min(len))
}

impl Wal {
    /// Open a WAL handle that can never write, for the read-only snapshot-serving
    /// path. The file is opened read-only (no create, no append, no truncate) and
    /// the handle carries no writer; every append/flush/commit path is unreachable
    /// in read-only engine mode. Record reads still work because they reopen the
    /// file read-only on demand.
    pub fn open_read_only(path: &Path, batch_size: usize) -> io::Result<Self> {
        let records_start = if path.exists() {
            let mut f = File::open(path)?;
            wal_records_start_readonly(&mut f)?
        } else {
            WAL_FILE_HEADER_SIZE
        };
        Ok(Wal {
            path: path.to_path_buf(),
            writer: None,
            batch_size,
            pending: 0,
            sync_mode: WalSyncMode::Off,
            next_lsn: 1,
            records_start,
            synced_len: records_start,
            shared: Arc::new(WalSyncShared::new(None)),
            defer_sync: false,
            deferred_gen: None,
            flusher: None,
        })
    }

    /// Truncate the WAL (after checkpoint).
    pub fn truncate(&mut self) -> io::Result<()> {
        // Settle any deferred durability claim before destroying the records
        // it covers: this keeps the "WAL records are durable before truncate"
        // ordering airtight even if a caller checkpoints while deferral is
        // active.
        if let Some(gen) = self.deferred_gen.take() {
            self.shared.sync_until(gen)?;
        }
        let mut file = OpenOptions::new()
            .write(true)
            .read(true)
            .truncate(true)
            .open(&self.path)?;
        write_wal_file_header(&mut file)?;
        let sync_fd = file.try_clone()?;
        self.writer = Some(BufWriter::new(file));
        // The old records are gone; settle their generations and swap the
        // fsync fd so outstanding tickets can never block on them.
        self.shared.replace_file(Some(sync_fd));
        self.records_start = WAL_FILE_HEADER_SIZE;
        self.pending = 0;
        self.synced_len = WAL_FILE_HEADER_SIZE;
        Ok(())
    }

    /// Discard records appended since the last successful [`Self::flush`].
    ///
    /// This is intentionally different from `flush`: it must not flush the
    /// current `BufWriter`, because rollback uses it to abandon uncommitted
    /// transaction records. `BufWriter::into_parts` lets us drop the buffered
    /// bytes without writing them, then we truncate any large records that
    /// had already spilled through to the file back to the last synced
    /// boundary.
    pub fn discard_pending(&mut self) -> io::Result<()> {
        if matches!(self.sync_mode, WalSyncMode::Off) {
            self.pending = 0;
            return Ok(());
        }

        if let Some(writer) = self.writer.take() {
            let (_file, _buffer) = writer.into_parts();
        }

        let file = OpenOptions::new()
            .read(true)
            .append(true)
            .create(true)
            .truncate(false)
            .open(&self.path)?;
        file.set_len(self.synced_len)?;
        file.sync_data()?;
        let sync_fd = file.try_clone()?;
        self.writer = Some(BufWriter::new(file));
        // The surviving prefix was just `sync_data`ed; settle all registered
        // generations and drop any deferred claim over discarded bytes.
        self.deferred_gen = None;
        self.shared.replace_file(Some(sync_fd));
        self.pending = 0;
        self.synced_len = self.records_start;
        Ok(())
    }
}

impl Drop for Wal {
    fn drop(&mut self) {
        // Clean shutdown must be durable regardless of mode: push any buffered
        // bytes to the OS and fsync, so a Normal-mode commit that hasn't yet
        // hit the background flusher's interval is still durable on a graceful
        // exit. (A process *crash* skips this — Normal's bounded-loss contract
        // only applies to OS-crash / power-loss, which this cannot help.)
        if !matches!(self.sync_mode, WalSyncMode::Off) {
            if let Some(writer) = self.writer.as_mut() {
                let _ = writer.flush();
                let _ = writer.get_ref().sync_data();
            }
        }
        self.stop_flusher();
        #[cfg(any(test, feature = "testing"))]
        self.disarm_fsync_failpoint();
    }
}

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

    fn temp_wal(name: &str) -> (Wal, PathBuf) {
        let path = std::env::temp_dir().join(format!("powdb_wal_{name}_{}", std::process::id()));
        let wal = Wal::create(&path, 4).unwrap();
        (wal, path)
    }

    #[test]
    fn test_append_and_flush() {
        let (mut wal, path) = temp_wal("basic");
        wal.append(1, WalRecordType::Insert, b"row data 1").unwrap();
        wal.append(1, WalRecordType::Insert, b"row data 2").unwrap();
        wal.flush().unwrap();

        let records = wal.read_all().unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].tx_id, 1);
        assert_eq!(records[0].data, b"row data 1");
        assert_eq!(records[1].data, b"row data 2");
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_group_commit_auto_flush() {
        let (mut wal, path) = temp_wal("group");
        // Batch size is 4 — after 4 appends, should auto-flush
        for i in 0..4 {
            wal.append(1, WalRecordType::Insert, format!("row {i}").as_bytes())
                .unwrap();
        }
        // Should have flushed automatically
        let records = wal.read_all().unwrap();
        assert_eq!(records.len(), 4);
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_normal_mode_persists_records_across_reopen() {
        // NORMAL durability: commits are acked after the buffered bytes reach
        // the OS (BufWriter::flush) without a per-commit fsync; a background
        // flusher + clean shutdown make them durable. Data must survive a
        // clean close + reopen.
        let path =
            std::env::temp_dir().join(format!("powdb_wal_normal_reopen_{}", std::process::id()));
        std::fs::remove_file(&path).ok();
        {
            let mut wal = Wal::create(&path, 4).unwrap();
            wal.set_sync_mode(WalSyncMode::Normal);
            assert_eq!(wal.sync_mode(), WalSyncMode::Normal);
            wal.append(1, WalRecordType::Insert, b"n1").unwrap();
            wal.append(1, WalRecordType::Insert, b"n2").unwrap();
            wal.flush().unwrap();
        } // drop: stop flusher + final fsync
        let wal = Wal::open(&path, 4).unwrap();
        let records = wal.read_all().unwrap();
        assert_eq!(records.len(), 2);
        assert_eq!(records[0].data, b"n1");
        assert_eq!(records[1].data, b"n2");
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_normal_mode_background_flusher_syncs_off_commit_path() {
        // In NORMAL mode flush() must NOT fsync inline; the background flusher
        // fsyncs on its interval and advances the synced generation. Proves the
        // fsync is off the commit path (the latency win) yet still happens.
        let path = std::env::temp_dir().join(format!("powdb_wal_normal_bg_{}", std::process::id()));
        std::fs::remove_file(&path).ok();
        let mut wal = Wal::create(&path, 1000).unwrap(); // large batch: no auto-flush
        wal.set_sync_mode(WalSyncMode::Normal);
        wal.append(1, WalRecordType::Insert, b"bg1").unwrap();
        wal.flush().unwrap(); // buffers to OS + marks dirty; no inline fsync
                              // The background flusher fsyncs on its (~10 ms) interval. Poll
                              // rather than sleeping a fixed 80 ms: on loaded CI runners the
                              // flusher thread can be starved well past one interval, and the
                              // property under test is "it happens off the commit path", not
                              // "it happens within 80 ms".
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
        while wal.synced_generation() < 1 && std::time::Instant::now() < deadline {
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(
            wal.synced_generation() >= 1,
            "background flusher did not sync within 2s (synced_generation = {})",
            wal.synced_generation()
        );
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_lone_committer_fsyncs_immediately_per_commit() {
        // Group commit must never delay a lone committer: with no other
        // waiters, every flush fsyncs immediately — exactly one fsync per
        // commit, no timers, no batching window.
        let (mut wal, path) = temp_wal("lone_committer");
        let base = wal.fsync_count();
        for i in 0..10u32 {
            wal.append(1, WalRecordType::Insert, format!("c{i}").as_bytes())
                .unwrap();
            wal.flush().unwrap();
        }
        assert_eq!(
            wal.fsync_count() - base,
            10,
            "a lone sequential committer must fsync exactly once per commit"
        );
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_deferred_tickets_coalesce_one_fsync_for_two_commits() {
        // Two commits registered before either waits: the first wait's fsync
        // covers both generations, the second wait returns without an fsync.
        let path = std::env::temp_dir().join(format!(
            "powdb_wal_gc_coalesce2_{}_{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let mut wal = Wal::create(&path, 1024).unwrap();
        wal.set_defer_sync(true);

        wal.append(1, WalRecordType::Insert, b"a").unwrap();
        wal.flush().unwrap();
        let t1 = wal.take_durability_ticket().expect("ticket for commit 1");

        wal.append(2, WalRecordType::Insert, b"b").unwrap();
        wal.flush().unwrap();
        let t2 = wal.take_durability_ticket().expect("ticket for commit 2");

        let base = wal.fsync_count();
        t2.wait().unwrap(); // leader — its fsync covers generation 1 too
        t1.wait().unwrap(); // already covered, no second fsync
        assert_eq!(
            wal.fsync_count() - base,
            1,
            "one fsync must cover both queued commits"
        );
        assert_eq!(wal.read_all().unwrap().len(), 2);
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_concurrent_committers_share_one_fsync() {
        // Classic group commit: N committers append + register (serialized by
        // the writer lock), all reach the barrier before any of them waits,
        // then the first waiter's fsync covers every registered generation.
        use std::sync::Barrier;

        let path = std::env::temp_dir().join(format!(
            "powdb_wal_gc_concurrent_{}_{}",
            std::process::id(),
            std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_nanos()
        ));
        let wal = Arc::new(Mutex::new(Wal::create(&path, 1024).unwrap()));
        wal.lock().unwrap().set_defer_sync(true);

        let n = 8;
        let barrier = Arc::new(Barrier::new(n));
        let mut handles = Vec::new();
        for t in 0..n {
            let wal = Arc::clone(&wal);
            let barrier = Arc::clone(&barrier);
            handles.push(std::thread::spawn(move || {
                let ticket = {
                    let mut w = wal.lock().unwrap();
                    w.append(t as u64 + 1, WalRecordType::Insert, b"row")
                        .unwrap();
                    w.flush().unwrap();
                    w.take_durability_ticket().expect("deferred ticket")
                };
                barrier.wait();
                ticket.wait().unwrap();
            }));
        }
        for h in handles {
            h.join().unwrap();
        }

        let w = wal.lock().unwrap();
        assert_eq!(w.read_all().unwrap().len(), n);
        assert_eq!(
            w.fsync_count(),
            1,
            "all {n} overlapping commits must be covered by a single fsync"
        );
        drop(w);
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    /// Byte offset of the `index`-th record in a WAL file, walking the length
    /// fields from the file header.
    fn record_offset(bytes: &[u8], index: usize) -> usize {
        let mut offset = WAL_FILE_HEADER_SIZE as usize;
        for _ in 0..index {
            let len = u32::from_le_bytes(
                bytes[offset..offset + 4]
                    .try_into()
                    .expect("4-byte length field"),
            ) as usize;
            offset += len;
        }
        offset
    }

    fn three_record_wal(name: &str) -> (Wal, PathBuf) {
        let (mut wal, path) = temp_wal(name);
        for i in 1..=3u64 {
            wal.append(i, WalRecordType::Insert, format!("record-{i}").as_bytes())
                .unwrap();
        }
        wal.flush().unwrap();
        (wal, path)
    }

    #[test]
    fn a_clean_log_reads_every_record_back() {
        let (wal, path) = three_record_wal("crc_clean");
        let records = wal.read_all().unwrap();
        assert_eq!(records.len(), 3);
        assert_eq!(records[0].data, b"record-1");
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    /// Damage in the MIDDLE of the log is not a torn tail: the records after it
    /// were written and committed. Truncating there silently discarded them.
    #[test]
    fn a_corrupt_record_in_the_middle_of_the_log_refuses_replay() {
        let (wal, path) = three_record_wal("crc_middle");
        drop(wal);

        let mut bytes = std::fs::read(&path).unwrap();
        let second = record_offset(&bytes, 1);
        bytes[second + 4] ^= 0xFF; // flip the stored CRC of record 2
        std::fs::write(&path, &bytes).unwrap();

        let wal = Wal::open(&path, 128).unwrap();
        let error = wal
            .read_all()
            .expect_err("mid-log damage must refuse, not truncate");
        assert_eq!(
            crate::error::StorageError::kind_of_io_error(&error),
            Some(crate::error::StorageErrorKind::WalReplay),
            "expected a WAL replay refusal, got: {error}"
        );
        assert!(
            error.to_string().contains("damage inside the log"),
            "the refusal must say why it is not a torn tail, got: {error}"
        );
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    /// Damage in the LAST record is the torn tail of an interrupted write.
    /// Everything before it is intact and replayable.
    #[test]
    fn a_corrupt_record_at_the_tail_is_discarded_as_a_torn_write() {
        let (wal, path) = three_record_wal("crc_tail");
        drop(wal);

        let mut bytes = std::fs::read(&path).unwrap();
        let third = record_offset(&bytes, 2);
        bytes[third + 4] ^= 0xFF; // flip the stored CRC of the last record
        std::fs::write(&path, &bytes).unwrap();

        let wal = Wal::open(&path, 128).unwrap();
        let records = wal.read_all().expect("a torn tail is recoverable");
        assert_eq!(records.len(), 2, "the two intact records must survive");
        assert_eq!(records[1].data, b"record-2");
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_multiple_transactions() {
        let (mut wal, path) = temp_wal("multi_tx");
        wal.append(1, WalRecordType::Insert, b"tx1 op1").unwrap();
        wal.append(2, WalRecordType::Insert, b"tx2 op1").unwrap();
        wal.append(1, WalRecordType::Commit, b"").unwrap();
        wal.append(2, WalRecordType::Commit, b"").unwrap();
        wal.flush().unwrap();

        let records = wal.read_all().unwrap();
        assert_eq!(records.len(), 4);
        assert_eq!(records[0].tx_id, 1);
        assert_eq!(records[2].tx_id, 1);
        assert_eq!(records[2].record_type, WalRecordType::Commit);
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_overflow_record_types_roundtrip() {
        // Additive record types 11/12 append, flush, and read back with their
        // type + payload intact, alongside the existing Insert/Commit records.
        let (mut wal, path) = temp_wal("ovf_types");
        wal.append(1, WalRecordType::OverflowWrite, b"chunk-payload")
            .unwrap();
        wal.append(1, WalRecordType::OverflowFree, b"\x02\x00\x00\x00")
            .unwrap();
        wal.append(1, WalRecordType::Insert, b"stub-row").unwrap();
        wal.append(1, WalRecordType::Commit, b"").unwrap();
        wal.flush().unwrap();

        let records = wal.read_all().unwrap();
        assert_eq!(records.len(), 4);
        assert_eq!(records[0].record_type, WalRecordType::OverflowWrite);
        assert_eq!(records[0].data, b"chunk-payload");
        assert_eq!(records[1].record_type, WalRecordType::OverflowFree);
        assert_eq!(records[2].record_type, WalRecordType::Insert);
        assert_eq!(records[3].record_type, WalRecordType::Commit);
        assert_eq!(
            WalRecordType::from_u8(11),
            Some(WalRecordType::OverflowWrite)
        );
        assert_eq!(
            WalRecordType::from_u8(12),
            Some(WalRecordType::OverflowFree)
        );
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_truncate() {
        let (mut wal, path) = temp_wal("trunc");
        for i in 0..8 {
            wal.append(1, WalRecordType::Insert, format!("data {i}").as_bytes())
                .unwrap();
        }
        wal.flush().unwrap();
        assert_eq!(wal.read_all().unwrap().len(), 8);

        wal.truncate().unwrap();
        assert_eq!(wal.read_all().unwrap().len(), 0);
        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    #[test]
    fn test_reopen_wal() {
        let path = std::env::temp_dir().join(format!("powdb_wal_reopen_{}", std::process::id()));
        {
            let mut wal = Wal::create(&path, 128).unwrap();
            wal.append(1, WalRecordType::Insert, b"persistent").unwrap();
            wal.append(1, WalRecordType::Commit, b"").unwrap();
            wal.flush().unwrap();
        }
        {
            let wal = Wal::open(&path, 128).unwrap();
            let records = wal.read_all().unwrap();
            assert_eq!(records.len(), 2);
            assert_eq!(records[0].data, b"persistent");
            assert_eq!(records[1].record_type, WalRecordType::Commit);
        }
        std::fs::remove_file(&path).ok();
    }

    /// fsyncgate: after a failed fsync, the OS may have dropped the dirty
    /// pages and marked them clean — a RETRIED fsync then reports success
    /// without the data ever reaching stable storage. The only sound
    /// response is to poison the WAL: surface the failure once, then refuse
    /// every later durability claim until the process restarts and replays
    /// the on-disk log.
    #[test]
    fn a_failed_fsync_poisons_the_wal_and_is_never_retried() {
        let (mut wal, path) = temp_wal("fsync_poison");
        wal.set_sync_mode(WalSyncMode::Full);
        wal.append(1, WalRecordType::Insert, b"healthy").unwrap();
        wal.flush().unwrap();
        let healthy_fsyncs = wal.fsync_count();
        assert!(healthy_fsyncs >= 1, "baseline commit must fsync");
        assert!(!wal.is_sync_poisoned());

        wal.arm_fsync_failpoint();
        wal.append(1, WalRecordType::Insert, b"doomed").unwrap();
        let err = wal
            .flush()
            .expect_err("the injected fsync failure must surface");
        assert!(
            err.to_string().contains("injected"),
            "first failure surfaces the real error, got: {err}"
        );
        assert!(wal.is_sync_poisoned());

        // The failpoint disarmed itself, so a retried fsync WOULD report
        // success now — exactly the false durability this test forbids.
        wal.append(1, WalRecordType::Insert, b"after").unwrap();
        let err = wal
            .flush()
            .expect_err("a poisoned WAL must refuse further durability claims");
        assert!(
            err.to_string().contains("poisoned"),
            "later failures name the poisoned state, got: {err}"
        );
        assert_eq!(
            wal.fsync_count(),
            healthy_fsyncs,
            "no fsync may ever be issued after a failure"
        );

        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    /// Same property through the Normal-mode background flusher: its fsync
    /// failing must poison the WAL (stopping the flusher) rather than retry
    /// on the next tick, and later writes must fail loudly instead of being
    /// acked against a durability mechanism that no longer exists.
    #[test]
    fn a_failed_background_fsync_poisons_normal_mode() {
        let (mut wal, path) = temp_wal("flusher_poison");
        wal.set_sync_mode(WalSyncMode::Normal);

        wal.arm_fsync_failpoint();
        wal.append(1, WalRecordType::Insert, b"doomed").unwrap();
        wal.flush().unwrap(); // Normal: flush-to-OS only, fsync is the flusher's

        // The flusher ticks every ~10ms; poll generously for the poison.
        let deadline = std::time::Instant::now() + Duration::from_secs(10);
        while !wal.is_sync_poisoned() && std::time::Instant::now() < deadline {
            std::thread::sleep(Duration::from_millis(5));
        }
        assert!(wal.is_sync_poisoned(), "flusher fsync failure must poison");

        let synced_after_poison = wal.synced_generation();
        wal.append(1, WalRecordType::Insert, b"after").unwrap();
        let err = wal
            .flush()
            .expect_err("poisoned Normal-mode WAL must refuse further commits");
        assert!(err.to_string().contains("poisoned"), "got: {err}");
        // Disarmed failpoint + still-running flusher would falsely advance
        // the synced generation; the poisoned WAL must never move it again.
        std::thread::sleep(Duration::from_millis(50));
        assert_eq!(wal.synced_generation(), synced_after_poison);

        drop(wal);
        std::fs::remove_file(&path).ok();
    }

    /// Two tests arming their own WALs concurrently must not steal each
    /// other's injection: a single-slot failpoint let the second arm clobber
    /// the first, so the first WAL's fsync silently succeeded (the CI-only
    /// flake where both fsyncgate tests started in the same instant).
    #[test]
    fn arming_two_wals_keeps_both_failpoints_armed() {
        let (mut wal_a, path_a) = temp_wal("failpoint_two_a");
        let (mut wal_b, path_b) = temp_wal("failpoint_two_b");
        wal_a.set_sync_mode(WalSyncMode::Full);
        wal_b.set_sync_mode(WalSyncMode::Full);

        wal_a.arm_fsync_failpoint();
        wal_b.arm_fsync_failpoint();

        wal_a.append(1, WalRecordType::Insert, b"a").unwrap();
        wal_a
            .flush()
            .expect_err("the first-armed WAL must still see its injected failure");
        wal_b.append(1, WalRecordType::Insert, b"b").unwrap();
        wal_b
            .flush()
            .expect_err("the second-armed WAL must see its injected failure too");

        drop(wal_a);
        drop(wal_b);
        std::fs::remove_file(&path_a).ok();
        std::fs::remove_file(&path_b).ok();
    }

    /// A WAL that dies with its failpoint still armed must take the arm with
    /// it. A stale armed address would otherwise match a later WAL that the
    /// allocator happens to place at the same spot, injecting an fsync
    /// failure into an unrelated test.
    #[test]
    fn dropping_an_armed_wal_disarms_its_failpoint() {
        let (wal, path) = temp_wal("failpoint_drop");
        // Holding this clone keeps the shared allocation alive across the
        // drop, so no concurrent test's WAL can reoccupy the address and
        // re-arm it — the containment checks below race with nothing.
        let shared = Arc::clone(&wal.shared);
        let addr = Arc::as_ptr(&shared) as usize;

        wal.arm_fsync_failpoint();
        assert!(WAL_FSYNC_FAILPOINTS.lock().unwrap().contains(&addr));
        drop(wal);
        assert!(
            !WAL_FSYNC_FAILPOINTS.lock().unwrap().contains(&addr),
            "dropping the WAL must remove its armed failpoint"
        );
        std::fs::remove_file(&path).ok();
    }
}