sochdb_storage/

txn_wal.rs

// SPDX-License-Identifier: AGPL-3.0-or-later
// SochDB - LLM-Optimized Embedded Database
// Copyright (C) 2026 Sushanth Reddy Vanagala (https://github.com/sushanthpy)
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published by
// the Free Software Foundation, either version 3 of the License, or
// (at your option) any later version.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.

//! Transaction-Aware WAL for ACID Transactions
//!
//! This WAL implementation provides ACID transaction support:
//! - Atomicity: All writes in a transaction are logged together
//! - Consistency: Schema validation before commit
//! - Isolation: Transaction IDs for MVCC
//! - Durability: fsync after commit
//!
//! ## WAL Record Types
//!
//! - Data: Key-value write
//! - TxnBegin: Start of transaction
//! - TxnCommit: Commit point (durability guarantee)
//! - TxnAbort: Transaction rollback
//! - Checkpoint: Snapshot marker for recovery
//! - SchemaChange: DDL operations
//!
//! ## Record Format (Fixed Layout for Torn Write Detection)
//!
//! ```text
//! ┌───────────┬──────────┬─────────┬───────────┬─────────┬───────────┬─────────┬─────────┬──────────┐
//! │ Length(4) │ Type(1)  │ TxnId(8)│ Timestamp │ KeyLen  │ ValueLen  │ Key(*)  │ Value(*)│ CRC32(4) │
//! │           │          │         │   (8)     │  (4)    │   (4)     │         │         │          │
//! └───────────┴──────────┴─────────┴───────────┴─────────┴───────────┴─────────┴─────────┴──────────┘
//! ```
//!
//! ## Crash Recovery Guarantees
//!
//! 1. **Torn Write Detection**: Length prefix + CRC32 checksum detects partial writes
//! 2. **Atomic Commit**: Commit record with fsync ensures durability
//! 3. **Uncommitted Rollback**: Transactions without commit record are discarded
//! 4. **Checkpoint Safety**: Checkpoint marker allows safe WAL truncation

use crate::encryption::EncryptionEngine;
use byteorder::{LittleEndian, ReadBytesExt};
use parking_lot::Mutex;
use sochdb_core::{Result, SochDBError, WalRecordType};
use std::cell::Cell;
use std::collections::HashSet;
use std::fs::{File, OpenOptions};
use std::io::{BufReader, BufWriter, Read, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::{AtomicU64, Ordering};
use std::time::{Instant, SystemTime, UNIX_EPOCH};

// =============================================================================
// Coarse-Grained Timestamp Caching (Recommendation 5)
// =============================================================================

/// Cache validity period in nanoseconds (1ms - allows ~1000+ writes per refresh)
const CACHE_VALIDITY_NS: u64 = 1_000_000;

thread_local! {
    /// Thread-local timestamp cache: (Instant, cached_timestamp_us)
    /// Avoids syscall overhead by caching timestamp for ~1ms windows
    static TS_CACHE: Cell<(Instant, u64)> = Cell::new((Instant::now(), 0));
}

/// Get cached timestamp in microseconds
///
/// This function eliminates per-write `SystemTime::now()` syscalls by:
/// 1. Caching the wall-clock timestamp in thread-local storage
/// 2. Using monotonic `Instant` for sub-millisecond offsets
/// 3. Only refreshing from syscall when cache expires (every ~1ms)
///
/// ## Performance Impact
///
/// - Without caching: ~20-25ns per syscall × 1M writes = 20-25ms overhead
/// - With caching: ~20ns per 1000 writes = 0.02ms overhead (1000× improvement)
///
/// ## Monotonicity Guarantee
///
/// Timestamps are guaranteed to be monotonically increasing within a thread
/// due to the `elapsed` offset added to the cached base timestamp.
#[inline(always)]
pub fn cached_timestamp_us() -> u64 {
    TS_CACHE.with(|cache| {
        let (instant, ts) = cache.get();
        let elapsed_ns = instant.elapsed().as_nanos() as u64;

        if elapsed_ns < CACHE_VALIDITY_NS {
            // Fast path: return cached + monotonic offset
            // This is pure arithmetic, no syscall
            ts + elapsed_ns / 1000
        } else {
            // Slow path: refresh cache from syscall
            let new_ts = SystemTime::now()
                .duration_since(UNIX_EPOCH)
                .expect("system clock set before UNIX epoch (1970-01-01)")
                .as_micros() as u64;
            cache.set((Instant::now(), new_ts));
            new_ts
        }
    })
}

/// Header size in bytes (without key/value data)
const RECORD_HEADER_SIZE: usize = 4 + 1 + 8 + 8 + 4 + 4; // length + type + txn_id + timestamp + key_len + value_len

/// Checksum size (CRC32)
const CHECKSUM_SIZE: usize = 4;

/// Default capacity for transaction-local WAL buffer (32 KB - typical batch of 100-500 writes)
const DEFAULT_TXN_BUFFER_CAPACITY: usize = 32 * 1024;

// =============================================================================
// At-rest encryption framing (Task 3B)
// =============================================================================
//
// When the WAL's `EncryptionEngine` is enabled, each record is written as an
// encrypted frame instead of the plaintext frame above:
//
// ```text
//   plaintext frame:  [content_len:u32][type|txn|ts|klen|vlen|key|val|crc32]
//   encrypted frame:  [outer_len:u32  ][ver|nonce(12)|ciphertext+tag(16)]
// ```
//
// `outer_len` is the length of the AEAD envelope (ciphertext.len()); the inner
// plaintext that the envelope protects is the record BODY — exactly the bytes
// after `content_len` in the plaintext frame (`type..crc32`). On decrypt the
// body is parsed by [`TxnWalEntry::parse_body`], reusing the same field layout +
// CRC check as the plaintext path.
//
// The per-record AAD binds {format_version, db_uuid, dek_epoch, file-relative
// record ordinal} so a record cannot be reordered, duplicated, spliced from
// another DB, or read under a downgraded format without failing authentication.
// The reader reconstructs the identical AAD from its own trusted state (the
// keyring's db_uuid/epoch and a 0-based counter of records read from this file),
// NEVER from the attacker-controllable on-disk bytes.

/// Encode a record BODY (`type|txn|ts|klen|vlen|key|val|crc32`) — the bytes the
/// AEAD envelope protects, identical to `to_bytes()` minus the length prefix.
/// Used by the encrypted write paths to materialize a record before sealing it.
fn encode_record_body(
    record_type: WalRecordType,
    txn_id: u64,
    timestamp_us: u64,
    key: &[u8],
    value: &[u8],
) -> Vec<u8> {
    let body_len = (RECORD_HEADER_SIZE - 4) + key.len() + value.len() + CHECKSUM_SIZE;
    let mut body = Vec::with_capacity(body_len);
    let mut hasher = crc32fast::Hasher::new();
    let rt = record_type as u8;
    body.push(rt);
    hasher.update(&[rt]);
    let t = txn_id.to_le_bytes();
    body.extend_from_slice(&t);
    hasher.update(&t);
    let ts = timestamp_us.to_le_bytes();
    body.extend_from_slice(&ts);
    hasher.update(&ts);
    let kl = (key.len() as u32).to_le_bytes();
    body.extend_from_slice(&kl);
    hasher.update(&kl);
    let vl = (value.len() as u32).to_le_bytes();
    body.extend_from_slice(&vl);
    hasher.update(&vl);
    body.extend_from_slice(key);
    hasher.update(key);
    body.extend_from_slice(value);
    hasher.update(value);
    body.extend_from_slice(&hasher.finalize().to_le_bytes());
    body
}

/// A Point-in-Time Recovery target (Task 3B PITR). See
/// [`TxnWal::replay_to_target`] for the prefix semantics.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RecoveryTarget {
    /// Recover the first `lsn` WAL records (exact). Matches the value of
    /// `DurableStorage::current_lsn()` captured at the desired point.
    Lsn(u64),
    /// Roll the WAL forward and STOP at the first transaction whose COMMIT
    /// timestamp (microseconds since the epoch) exceeds this value — that
    /// transaction and everything after it are excluded. This is an exact
    /// WAL-order PREFIX, NOT a timestamp filter: if commit timestamps are
    /// non-monotonic in WAL order (coarse ~1ms clock, NTP steps, cross-thread
    /// group commit), a later-in-WAL transaction with `ts <= t` that follows an
    /// excluded commit is ALSO excluded. Prefer `Lsn` for an exact, clock-
    /// independent cut.
    Timestamp(u64),
}

/// AAD layout version for the WAL record binding. Part of the on-disk contract.
const WAL_AAD_VERSION: u8 = 1;
/// AAD length: version(1) + db_uuid(16) + dek_epoch(4) + record_ordinal(8).
const WAL_AAD_LEN: usize = 1 + 16 + 4 + 8;
/// Upper bound on a single WAL frame's on-disk length, to reject a corrupted
/// length prefix before it triggers a multi-GB `read_exact` allocation/DoS.
const MAX_WAL_FRAME_LEN: u32 = 512 * 1024 * 1024;

// =============================================================================
// Transaction-Local WAL Buffer - Zero Lock Overhead During Transaction
// =============================================================================

/// Transaction-local WAL write buffer
///
/// Collects all writes during a transaction in memory, then flushes
/// everything with a SINGLE lock acquisition at commit time.
///
/// ## Performance Impact
///
/// ```text
/// Without TxnWalBuffer (current):
///   1000 writes × (lock + 9 write_all + unlock) = 1000 lock acquisitions
///   Lock overhead: 1000 × 20ns = 20µs per transaction
///
/// With TxnWalBuffer:
///   1000 writes buffered locally (NO LOCK)
///   1 flush at commit (SINGLE LOCK)
///   Lock overhead: 1 × 20ns = 20ns per transaction
///   Speedup: 1000× for lock overhead
/// ```
///
/// ## Usage
///
/// ```ignore
/// let mut buffer = TxnWalBuffer::new(txn_id);
///
/// // These do NOT acquire any locks
/// buffer.append(b"key1", b"value1");
/// buffer.append(b"key2", b"value2");
/// // ... hundreds of writes ...
///
/// // Single lock acquisition, single write syscall
/// buffer.flush(&wal)?;
/// ```
#[derive(Debug)]
pub struct TxnWalBuffer {
    /// Transaction ID for all entries
    txn_id: u64,
    /// Accumulated serialized entries
    buffer: Vec<u8>,
    /// Number of entries buffered
    entry_count: usize,
}

impl TxnWalBuffer {
    /// Create a new buffer for a transaction
    #[inline]
    pub fn new(txn_id: u64) -> Self {
        Self {
            txn_id,
            buffer: Vec::with_capacity(DEFAULT_TXN_BUFFER_CAPACITY),
            entry_count: 0,
        }
    }

    /// Create with specific capacity
    #[inline]
    pub fn with_capacity(txn_id: u64, capacity: usize) -> Self {
        Self {
            txn_id,
            buffer: Vec::with_capacity(capacity),
            entry_count: 0,
        }
    }

    /// Append a key-value write to the buffer - NO LOCK, NO SYSCALL
    ///
    /// Serializes the WAL entry directly to the buffer with CRC32 calculation.
    /// This is completely lock-free and does not touch the file system.
    ///
    /// Uses cached timestamps to eliminate per-write syscall overhead.
    #[inline]
    pub fn append(&mut self, key: &[u8], value: &[u8]) {
        // Use cached timestamp instead of syscall (Recommendation 5)
        let timestamp_us = cached_timestamp_us();

        let total_len = RECORD_HEADER_SIZE + key.len() + value.len() + CHECKSUM_SIZE;
        let entry_start = self.buffer.len();

        // Reserve space for length prefix (will fill at end)
        self.buffer.extend_from_slice(&[0u8; 4]);

        let mut hasher = crc32fast::Hasher::new();

        // Record type (Data = 0)
        let record_type_byte = WalRecordType::Data as u8;
        self.buffer.push(record_type_byte);
        hasher.update(&[record_type_byte]);

        // Transaction ID
        let txn_bytes = self.txn_id.to_le_bytes();
        self.buffer.extend_from_slice(&txn_bytes);
        hasher.update(&txn_bytes);

        // Timestamp
        let ts_bytes = timestamp_us.to_le_bytes();
        self.buffer.extend_from_slice(&ts_bytes);
        hasher.update(&ts_bytes);

        // Key length
        let key_len_bytes = (key.len() as u32).to_le_bytes();
        self.buffer.extend_from_slice(&key_len_bytes);
        hasher.update(&key_len_bytes);

        // Value length
        let val_len_bytes = (value.len() as u32).to_le_bytes();
        self.buffer.extend_from_slice(&val_len_bytes);
        hasher.update(&val_len_bytes);

        // Key data
        self.buffer.extend_from_slice(key);
        hasher.update(key);

        // Value data
        self.buffer.extend_from_slice(value);
        hasher.update(value);

        // CRC32 checksum
        self.buffer
            .extend_from_slice(&hasher.finalize().to_le_bytes());

        // Fill in length prefix (content length, not including the 4-byte length field)
        let content_len = (total_len - 4) as u32;
        self.buffer[entry_start..entry_start + 4].copy_from_slice(&content_len.to_le_bytes());

        self.entry_count += 1;
    }

    /// Flush all buffered entries to WAL - SINGLE LOCK, SINGLE WRITE
    ///
    /// Acquires the WAL lock once and writes all accumulated entries.
    /// Returns the first sequence number assigned to buffered entries.
    ///
    /// Note: Use TxnWal::flush_buffer() instead of calling this directly,
    /// as it properly handles the private writer field.
    #[inline]
    pub fn flush_to_wal(&self, wal: &TxnWal) -> Result<u64> {
        wal.flush_buffer(self)
    }

    /// Clear the buffer (for reuse)
    #[inline]
    pub fn clear(&mut self) {
        self.buffer.clear();
        self.entry_count = 0;
    }

    /// Get number of buffered entries
    #[inline]
    pub fn entry_count(&self) -> usize {
        self.entry_count
    }

    /// Get total bytes buffered
    #[inline]
    pub fn bytes_buffered(&self) -> usize {
        self.buffer.len()
    }

    /// Check if buffer is empty
    #[inline]
    pub fn is_empty(&self) -> bool {
        self.buffer.is_empty()
    }
}

/// WAL entry for transaction-aware operations
#[derive(Debug, Clone)]
pub struct TxnWalEntry {
    /// Record type
    pub record_type: WalRecordType,
    /// Transaction ID
    pub txn_id: u64,
    /// Timestamp in microseconds
    pub timestamp_us: u64,
    /// Key data
    pub key: Vec<u8>,
    /// Value data
    pub value: Vec<u8>,
}

impl TxnWalEntry {
    /// Create a new data entry
    pub fn data(txn_id: u64, key: Vec<u8>, value: Vec<u8>) -> Self {
        Self {
            record_type: WalRecordType::Data,
            txn_id,
            timestamp_us: Self::now_us(),
            key,
            value,
        }
    }

    /// Create a transaction begin entry
    pub fn txn_begin(txn_id: u64) -> Self {
        Self {
            record_type: WalRecordType::TxnBegin,
            txn_id,
            timestamp_us: Self::now_us(),
            key: Vec::new(),
            value: Vec::new(),
        }
    }

    /// Create a transaction commit entry
    pub fn txn_commit(txn_id: u64) -> Self {
        Self {
            record_type: WalRecordType::TxnCommit,
            txn_id,
            timestamp_us: Self::now_us(),
            key: Vec::new(),
            value: Vec::new(),
        }
    }

    /// Create a transaction abort entry
    pub fn txn_abort(txn_id: u64) -> Self {
        Self {
            record_type: WalRecordType::TxnAbort,
            txn_id,
            timestamp_us: Self::now_us(),
            key: Vec::new(),
            value: Vec::new(),
        }
    }

    /// Create a checkpoint entry
    pub fn checkpoint(txn_id: u64) -> Self {
        Self {
            record_type: WalRecordType::Checkpoint,
            txn_id,
            timestamp_us: Self::now_us(),
            key: Vec::new(),
            value: Vec::new(),
        }
    }

    /// Create a schema change entry
    pub fn schema_change(txn_id: u64, schema_data: Vec<u8>) -> Self {
        Self {
            record_type: WalRecordType::SchemaChange,
            txn_id,
            timestamp_us: Self::now_us(),
            key: Vec::new(),
            value: schema_data,
        }
    }

    /// Get current time in microseconds (uses cached timestamp)
    #[inline]
    fn now_us() -> u64 {
        cached_timestamp_us()
    }

    /// Calculate CRC32 checksum for this entry
    ///
    /// Uses crc32fast for portable, deterministic checksums.
    /// The checksum covers all fields except the checksum itself.
    /// NOTE: This is only used for verification. to_bytes() calculates CRC in single pass.
    pub fn checksum(&self) -> u32 {
        let mut hasher = crc32fast::Hasher::new();
        hasher.update(&[self.record_type as u8]);
        hasher.update(&self.txn_id.to_le_bytes());
        hasher.update(&self.timestamp_us.to_le_bytes());
        hasher.update(&(self.key.len() as u32).to_le_bytes());
        hasher.update(&(self.value.len() as u32).to_le_bytes());
        hasher.update(&self.key);
        hasher.update(&self.value);
        hasher.finalize()
    }

    /// Serialize to bytes with single-pass CRC calculation
    ///
    /// Optimized to calculate CRC while building the buffer,
    /// avoiding a second pass over all data.
    pub fn to_bytes(&self) -> Vec<u8> {
        let total_len = RECORD_HEADER_SIZE + self.key.len() + self.value.len() + CHECKSUM_SIZE;
        let mut buf = Vec::with_capacity(total_len);
        let mut hasher = crc32fast::Hasher::new();

        // Length (not including the length field itself) - not included in CRC
        let content_len = (total_len - 4) as u32;
        buf.extend_from_slice(&content_len.to_le_bytes());

        // Record type
        let record_type_byte = self.record_type as u8;
        buf.push(record_type_byte);
        hasher.update(&[record_type_byte]);

        // Transaction ID
        let txn_bytes = self.txn_id.to_le_bytes();
        buf.extend_from_slice(&txn_bytes);
        hasher.update(&txn_bytes);

        // Timestamp
        let ts_bytes = self.timestamp_us.to_le_bytes();
        buf.extend_from_slice(&ts_bytes);
        hasher.update(&ts_bytes);

        // Key length
        let key_len_bytes = (self.key.len() as u32).to_le_bytes();
        buf.extend_from_slice(&key_len_bytes);
        hasher.update(&key_len_bytes);

        // Value length
        let val_len_bytes = (self.value.len() as u32).to_le_bytes();
        buf.extend_from_slice(&val_len_bytes);
        hasher.update(&val_len_bytes);

        // Key data
        buf.extend_from_slice(&self.key);
        hasher.update(&self.key);

        // Value data
        buf.extend_from_slice(&self.value);
        hasher.update(&self.value);

        // CRC32 Checksum (computed in single pass above)
        buf.extend_from_slice(&hasher.finalize().to_le_bytes());

        buf
    }

    /// Deserialize a PLAINTEXT WAL frame from a reader, with torn-write detection.
    ///
    /// This is the legacy / unencrypted reader. The live replay path on an
    /// encrypted WAL does NOT use this — it uses [`TxnWal::read_record`], which
    /// decrypts first and then calls [`Self::parse_body`]. Keeping this method
    /// plaintext-only ensures a stray caller can never silently mis-parse
    /// ciphertext as a plaintext frame.
    ///
    /// Returns error if:
    /// - Length field indicates data is too short or implausibly large
    /// - CRC32 checksum mismatch (corruption or torn write)
    /// - Invalid record type
    pub fn from_reader<R: Read>(reader: &mut R) -> Result<Self> {
        // Length (allows torn write detection - if length claims more data than exists)
        let content_len = reader.read_u32::<LittleEndian>()?;
        if content_len < (RECORD_HEADER_SIZE - 4 + CHECKSUM_SIZE) as u32 {
            return Err(SochDBError::Corruption("WAL entry too short".into()));
        }
        if content_len > MAX_WAL_FRAME_LEN {
            return Err(SochDBError::Corruption(format!(
                "WAL entry length {content_len} exceeds maximum {MAX_WAL_FRAME_LEN}"
            )));
        }

        // Read the whole body, then parse. A short read here (torn tail) surfaces
        // as Io(UnexpectedEof), which replay treats as a clean end-of-WAL.
        let mut body = vec![0u8; content_len as usize];
        reader.read_exact(&mut body)?;
        Self::parse_body(&body)
    }

    /// Parse a record BODY (`type|txn|ts|klen|vlen|key|val|crc32`) and verify its
    /// CRC. This is the bytes after the `content_len` prefix in a plaintext frame,
    /// and is exactly what the AEAD envelope protects in an encrypted frame — so
    /// both the plaintext and decrypt paths share this single parser.
    pub fn parse_body(body: &[u8]) -> Result<Self> {
        let mut cur = std::io::Cursor::new(body);

        let record_type_byte = cur.read_u8()?;
        let record_type = WalRecordType::try_from(record_type_byte).map_err(|_| {
            SochDBError::Corruption(format!("Invalid record type: {}", record_type_byte))
        })?;

        let txn_id = cur.read_u64::<LittleEndian>()?;
        let timestamp_us = cur.read_u64::<LittleEndian>()?;
        let key_len = cur.read_u32::<LittleEndian>()? as usize;
        let value_len = cur.read_u32::<LittleEndian>()? as usize;

        let mut key = vec![0u8; key_len];
        cur.read_exact(&mut key)?;
        let mut value = vec![0u8; value_len];
        cur.read_exact(&mut value)?;

        let stored_checksum = cur.read_u32::<LittleEndian>()?;

        let entry = Self {
            record_type,
            txn_id,
            timestamp_us,
            key,
            value,
        };

        // Verify checksum - detects both corruption and torn writes
        if entry.checksum() != stored_checksum {
            return Err(SochDBError::Corruption(format!(
                "WAL checksum mismatch for txn_id {}: expected {}, got {}",
                txn_id,
                entry.checksum(),
                stored_checksum
            )));
        }

        Ok(entry)
    }

    /// The record BODY bytes (`type..crc32`) — i.e. `to_bytes()` without the
    /// leading 4-byte content_len. This is what gets fed to the AEAD on the
    /// encrypted write path.
    fn body_bytes(&self) -> Vec<u8> {
        let full = self.to_bytes();
        full[4..].to_vec()
    }
}

/// Transaction-aware Write-Ahead Log
pub struct TxnWal {
    /// Path to WAL file
    path: PathBuf,
    /// Buffered writer
    writer: Mutex<BufWriter<File>>,
    /// Next transaction ID
    next_txn_id: AtomicU64,
    /// Write sequence number
    sequence: AtomicU64,
    /// Bytes written since last sync
    bytes_since_sync: AtomicU64,
    /// Cached timestamp (microseconds since epoch)
    /// Updated periodically to avoid syscall per write
    cached_timestamp_us: AtomicU64,
    /// At-rest encryption engine. `disabled()` (identity passthrough) for a
    /// plaintext WAL — in which case every write/read path is byte-identical to
    /// the pre-3B format. When enabled, records are written/read as encrypted
    /// frames (see the framing notes above).
    encryption: Arc<EncryptionEngine>,
    /// 16-byte database identity bound into each record's AAD (cross-DB splice
    /// defense). All-zero / unused when encryption is disabled.
    db_uuid: [u8; 16],
    /// Active DEK epoch bound into each record's AAD (downgrade-across-rotation
    /// defense). 0 / unused when encryption is disabled.
    dek_epoch: u32,
    /// File-relative count of records written to the CURRENT WAL file (reset on
    /// truncate). Used as the per-record AAD ordinal so reorder/duplication
    /// within a file fails authentication. All writes hold the `writer` lock, so
    /// this counter is only mutated under that lock; it is read locklessly by the
    /// (single-threaded) replay paths.
    records_in_file: AtomicU64,
}

impl TxnWal {
    /// Create a new plaintext WAL or open an existing one.
    ///
    /// Uses a disabled (passthrough) encryption engine, so the on-disk format is
    /// byte-identical to the pre-encryption WAL. This is the constructor used by
    /// every non-live caller (tests, tools); the live durable-storage path uses
    /// [`Self::new_with_encryption`] to thread the keyring's engine in.
    pub fn new<P: AsRef<Path>>(path: P) -> Result<Self> {
        Self::new_with_encryption(path, Arc::new(EncryptionEngine::disabled()), [0u8; 16], 0)
    }

    /// Create/open a WAL with an explicit at-rest encryption engine.
    ///
    /// When `engine` is enabled, records are written and replayed as encrypted
    /// frames bound to `db_uuid` + `dek_epoch` (see framing notes). When it is
    /// disabled, this behaves exactly like [`Self::new`].
    ///
    /// `pub(crate)` by design: the engine + `db_uuid` + `dek_epoch` MUST come
    /// from [`crate::keyring::load_or_init`] so the fail-closed open contract
    /// (MAC + canary) and the AAD binding are enforced. The only public door to
    /// an encrypted WAL is
    /// [`crate::durable_storage::DurableStorage::open_with_encryption`].
    pub(crate) fn new_with_encryption<P: AsRef<Path>>(
        path: P,
        encryption: Arc<EncryptionEngine>,
        db_uuid: [u8; 16],
        dek_epoch: u32,
    ) -> Result<Self> {
        let path = path.as_ref().to_path_buf();

        // Ensure parent directory exists
        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }

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

        // Use 256KB buffer for better batch performance (default is 8KB)
        // This reduces system calls when buffering many small writes
        let now_us = cached_timestamp_us();

        let wal = Self {
            path,
            writer: Mutex::new(BufWriter::with_capacity(256 * 1024, file)),
            next_txn_id: AtomicU64::new(1),
            sequence: AtomicU64::new(0),
            bytes_since_sync: AtomicU64::new(0),
            cached_timestamp_us: AtomicU64::new(now_us),
            encryption,
            db_uuid,
            dek_epoch,
            records_in_file: AtomicU64::new(0),
        };

        // Recover state from existing WAL
        wal.recover_state()?;

        Ok(wal)
    }

    /// Build the per-record AAD for a given file-relative ordinal. The reader
    /// reconstructs this from its own trusted state, never from on-disk bytes.
    #[inline]
    fn record_aad(&self, ordinal: u64) -> [u8; WAL_AAD_LEN] {
        let mut aad = [0u8; WAL_AAD_LEN];
        aad[0] = WAL_AAD_VERSION;
        aad[1..17].copy_from_slice(&self.db_uuid);
        aad[17..21].copy_from_slice(&self.dek_epoch.to_le_bytes());
        aad[21..29].copy_from_slice(&ordinal.to_le_bytes());
        aad
    }

    /// Frame a single record body for the encrypted write path:
    /// `[outer_len:u32][version|nonce|ciphertext+tag]`, AAD-bound to `ordinal`.
    #[inline]
    fn encrypt_frame(&self, body: &[u8], ordinal: u64) -> Result<Vec<u8>> {
        let env = self
            .encryption
            .encrypt_with_aad(body, &self.record_aad(ordinal))?;
        let mut out = Vec::with_capacity(4 + env.len());
        out.extend_from_slice(&(env.len() as u32).to_le_bytes());
        out.extend_from_slice(&env);
        Ok(out)
    }

    /// Read the next record from a replay reader, decrypting if the WAL is
    /// encrypted. Returns the record and advances `*ordinal`.
    ///
    /// Error taxonomy (the linchpin of fail-loud recovery):
    /// - `Io(UnexpectedEof)` reading the length prefix or a short body read at
    ///   the physical tail ⇒ clean torn-tail / end-of-WAL (callers tolerate).
    /// - `Encryption(_)` (AEAD auth failure / wrong key / bad envelope) or
    ///   `Corruption(_)` (CRC / bad framing) ⇒ HARD error; callers MUST abort
    ///   recovery rather than treat it as EOF.
    fn read_record<R: Read>(&self, reader: &mut R, ordinal: &mut u64) -> Result<TxnWalEntry> {
        if !self.encryption.is_enabled() {
            let entry = TxnWalEntry::from_reader(reader)?;
            *ordinal += 1;
            return Ok(entry);
        }
        // Encrypted frame: [outer_len][envelope]
        let outer_len = reader.read_u32::<LittleEndian>()?; // EOF here = clean end
        if outer_len > MAX_WAL_FRAME_LEN {
            return Err(SochDBError::Corruption(format!(
                "encrypted WAL frame length {outer_len} exceeds maximum {MAX_WAL_FRAME_LEN}"
            )));
        }
        let mut env = vec![0u8; outer_len as usize];
        reader.read_exact(&mut env)?; // short read = torn tail (UnexpectedEof)
        // Decrypt with the AAD reconstructed for THIS ordinal. A wrong key, a
        // reordered/spliced record, or tampering fails here as Encryption(_),
        // which is a hard error — never a silent EOF.
        let body = self
            .encryption
            .decrypt_with_aad(&env, &self.record_aad(*ordinal))?;
        let entry = TxnWalEntry::parse_body(&body)?;
        *ordinal += 1;
        Ok(entry)
    }

    /// Recover state (next txn ID, sequence) from existing WAL
    ///
    /// To avoid txn_id collisions when multiple processes open the same
    /// WAL concurrently, we incorporate the PID into the starting txn_id.
    /// Format: upper 32 bits = PID, lower 32 bits = counter.
    /// This guarantees uniqueness across processes without coordination.
    fn recover_state(&self) -> Result<()> {
        let file = File::open(&self.path)?;
        let mut reader = BufReader::new(file);
        let mut count: u64 = 0;

        // Track the max counter (lower 32 bits) for OUR PID only.
        // Each process owns its own txn_id space: upper 32 bits = PID.
        // We must NOT use max_txn_id across ALL PIDs, because that would
        // place us into another PID's ID space and cause collisions.
        let our_pid = std::process::id() as u64;
        let pid_base = our_pid << 32;
        let mut max_our_counter: u64 = 0;
        let mut ordinal: u64 = 0;

        loop {
            match self.read_record(&mut reader, &mut ordinal) {
                Ok(entry) => {
                    count += 1;
                    // Only track counters from entries that belong to our PID
                    let entry_pid = entry.txn_id >> 32;
                    if entry_pid == our_pid {
                        let entry_counter = entry.txn_id & 0xFFFF_FFFF;
                        if entry_counter > max_our_counter {
                            max_our_counter = entry_counter;
                        }
                    }
                }
                Err(SochDBError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    break;
                }
                Err(e) => {
                    // Encrypted: a non-EOF error (AEAD auth failure / tamper /
                    // corruption) is NEVER a torn tail — fail loud rather than
                    // silently truncate. Plaintext: tolerate trailing corruption
                    // as an incomplete final write (legacy behavior).
                    if self.encryption.is_enabled() {
                        return Err(e);
                    }
                    break;
                }
            }
        }

        // Start from pid_base + (max counter we've seen for our PID + 1).
        // This ensures:
        //   1. Unique across processes (different PID → different upper 32 bits)
        //   2. Unique within this process even if PID is recycled (we skip
        //      over any counters already used by a previous process with our PID)
        let next_id = pid_base + max_our_counter + 1;

        self.next_txn_id.store(next_id, Ordering::SeqCst);
        self.sequence.store(count, Ordering::SeqCst);
        // The current file already holds `count` records, so the next encrypted
        // write must continue the AAD ordinal sequence from there.
        self.records_in_file.store(count, Ordering::SeqCst);

        Ok(())
    }

    /// Get cached timestamp, updating if stale (>1ms old)
    ///
    /// This avoids a syscall per write by caching the timestamp.
    /// For WAL purposes, ~1ms granularity is sufficient.
    #[inline]
    fn get_cached_timestamp(&self) -> u64 {
        // Fast path: use cached value (no syscall)
        let cached = self.cached_timestamp_us.load(Ordering::Relaxed);

        // Refresh occasionally (every ~1000 writes or when sequence wraps)
        // This is a very cheap check since sequence is incremented atomically anyway
        let seq = self.sequence.load(Ordering::Relaxed);
        if seq & 0x3FF == 0 {
            // Refresh every 1024 writes
            let now_us = cached_timestamp_us();
            self.cached_timestamp_us.store(now_us, Ordering::Relaxed);
            return now_us;
        }

        cached
    }

    /// Append an entry to the WAL
    ///
    /// Returns the sequence number of this write.
    ///
    /// # Durability Contract
    ///
    /// This method writes data to the BufWriter but does **NOT** fsync.
    /// The data is NOT durable until `flush()` + `sync()` are called.
    ///
    /// **For transaction commits, use `commit_transaction()` or `commit_durable()`**
    /// which enforce fsync. This method is intentionally non-durable for
    /// batching data writes within a transaction (pre-commit records).
    ///
    /// The group commit path (`EventDrivenGroupCommit`) is responsible for
    /// calling `flush()` + `sync()` after batching multiple commit records.
    pub fn append(&self, entry: &TxnWalEntry) -> Result<u64> {
        let mut writer = self.writer.lock();
        let bytes = self.frame_for_write(entry)?;

        writer.write_all(&bytes)?;
        // Don't flush here - BufWriter will batch writes automatically.
        // Call flush() explicitly before sync() or commit().
        self.commit_record_ordinal();

        let seq = self.sequence.fetch_add(1, Ordering::SeqCst);
        self.bytes_since_sync
            .fetch_add(bytes.len() as u64, Ordering::Relaxed);

        Ok(seq)
    }

    /// Serialize one entry into its on-disk frame using the CURRENT (not-yet-
    /// advanced) AAD ordinal. MUST be called with the `writer` lock held, and the
    /// caller MUST call [`Self::commit_record_ordinal`] only AFTER `write_all`
    /// succeeds — so a failed or partial write never advances the in-memory
    /// ordinal past what is actually on disk (which would desync every
    /// subsequent record's AAD from the reader's reconstructed ordinal).
    #[inline]
    fn frame_for_write(&self, entry: &TxnWalEntry) -> Result<Vec<u8>> {
        if self.encryption.is_enabled() {
            let ord = self.records_in_file.load(Ordering::SeqCst);
            self.encrypt_frame(&entry.body_bytes(), ord)
        } else {
            Ok(entry.to_bytes())
        }
    }

    /// Advance the file-relative AAD ordinal by one record. Call ONLY after the
    /// record's bytes have been handed to the writer (post-`write_all`).
    #[inline]
    fn commit_record_ordinal(&self) {
        if self.encryption.is_enabled() {
            self.records_in_file.fetch_add(1, Ordering::SeqCst);
        }
    }

    /// Append entry without flushing (for batched writes)
    ///
    /// Caller must call flush() or sync() afterward to ensure durability.
    #[inline]
    pub fn append_no_flush(&self, entry: &TxnWalEntry) -> Result<u64> {
        let mut writer = self.writer.lock();
        let bytes = self.frame_for_write(entry)?;

        writer.write_all(&bytes)?;
        // No flush - let BufWriter buffer the writes
        self.commit_record_ordinal();

        let seq = self.sequence.fetch_add(1, Ordering::SeqCst);
        self.bytes_since_sync
            .fetch_add(bytes.len() as u64, Ordering::Relaxed);

        Ok(seq)
    }

    /// Write data without flushing (for batched writes within a transaction)
    #[inline]
    pub fn write_no_flush(&self, txn_id: u64, key: Vec<u8>, value: Vec<u8>) -> Result<u64> {
        let entry = TxnWalEntry::data(txn_id, key, value);
        self.append_no_flush(&entry)
    }

    /// Write data from slices without any allocation
    ///
    /// This is the fastest path for writing - no intermediate Vec allocations.
    /// Calculates CRC32 while serializing directly to BufWriter.
    #[inline]
    pub fn write_no_flush_refs(&self, txn_id: u64, key: &[u8], value: &[u8]) -> Result<u64> {
        // Use coarse timestamp (cached every ~1ms) instead of syscall per write
        let timestamp_us = self.get_cached_timestamp();

        let total_len = RECORD_HEADER_SIZE + key.len() + value.len() + CHECKSUM_SIZE;

        let mut writer = self.writer.lock();

        // Encrypted mode cannot stream field-by-field (the AEAD needs the whole
        // body to compute its synthetic IV + tag), so materialize the body into a
        // scratch buffer, seal it, and write one framed envelope. The plaintext
        // path below keeps its zero-alloc streaming fast path untouched.
        if self.encryption.is_enabled() {
            let body = encode_record_body(WalRecordType::Data, txn_id, timestamp_us, key, value);
            let ord = self.records_in_file.load(Ordering::SeqCst);
            let frame = self.encrypt_frame(&body, ord)?;
            writer.write_all(&frame)?;
            // Advance the AAD ordinal only after the bytes are committed to the
            // writer, so a failed write never desyncs writer/reader ordinals.
            self.records_in_file.fetch_add(1, Ordering::SeqCst);
            let seq = self.sequence.fetch_add(1, Ordering::SeqCst);
            self.bytes_since_sync
                .fetch_add(frame.len() as u64, Ordering::Relaxed);
            return Ok(seq);
        }

        let mut hasher = crc32fast::Hasher::new();

        // Length (not included in CRC)
        let content_len = (total_len - 4) as u32;
        writer.write_all(&content_len.to_le_bytes())?;

        // Record type
        let record_type_byte = WalRecordType::Data as u8;
        writer.write_all(&[record_type_byte])?;
        hasher.update(&[record_type_byte]);

        // Transaction ID
        let txn_bytes = txn_id.to_le_bytes();
        writer.write_all(&txn_bytes)?;
        hasher.update(&txn_bytes);

        // Timestamp
        let ts_bytes = timestamp_us.to_le_bytes();
        writer.write_all(&ts_bytes)?;
        hasher.update(&ts_bytes);

        // Key length
        let key_len_bytes = (key.len() as u32).to_le_bytes();
        writer.write_all(&key_len_bytes)?;
        hasher.update(&key_len_bytes);

        // Value length
        let val_len_bytes = (value.len() as u32).to_le_bytes();
        writer.write_all(&val_len_bytes)?;
        hasher.update(&val_len_bytes);

        // Key data
        writer.write_all(key)?;
        hasher.update(key);

        // Value data
        writer.write_all(value)?;
        hasher.update(value);

        // CRC32 checksum
        writer.write_all(&hasher.finalize().to_le_bytes())?;

        let seq = self.sequence.fetch_add(1, Ordering::SeqCst);
        self.bytes_since_sync
            .fetch_add(total_len as u64, Ordering::Relaxed);

        Ok(seq)
    }

    /// Flush pending writes to kernel buffer (but not to disk)
    pub fn flush(&self) -> Result<()> {
        let mut writer = self.writer.lock();
        writer.flush()?;
        Ok(())
    }

    /// Append and immediately sync for durability
    pub fn append_sync(&self, entry: &TxnWalEntry) -> Result<u64> {
        let seq = self.append(entry)?;
        self.sync()?;
        Ok(seq)
    }

    /// Force sync to disk.
    ///
    /// Must flush the BufWriter BEFORE fsync: `get_ref()` reaches the raw File
    /// and bypasses the 256 KB buffer, so without an explicit `flush()` the
    /// just-appended commit/checkpoint record may still sit in userspace and
    /// `sync_all()` would fsync stale bytes — silently breaking the durability
    /// guarantee `append_sync`/`commit`/`checkpoint` advertise.
    pub fn sync(&self) -> Result<()> {
        let mut writer = self.writer.lock();
        writer.flush()?;
        writer.get_ref().sync_all()?;
        self.bytes_since_sync.store(0, Ordering::Relaxed);
        Ok(())
    }

    /// Flush a TxnWalBuffer with single lock acquisition
    ///
    /// This is the high-performance path for transaction commit:
    /// - All writes during the transaction are buffered locally
    /// - At commit, this method flushes everything with ONE lock
    ///
    /// ## Performance
    ///
    /// ```text
    /// 1000 writes with individual flush: 1000 × lock overhead
    /// 1000 writes with buffer + flush_buffer: 1 × lock overhead
    /// Speedup: ~1000× for lock overhead
    /// ```
    #[inline]
    pub fn flush_buffer(&self, buffer: &TxnWalBuffer) -> Result<u64> {
        if buffer.is_empty() {
            return Ok(0);
        }

        let mut writer = self.writer.lock();

        if self.encryption.is_enabled() {
            // The buffer holds concatenated plaintext frames
            // `[content_len][body]...`. Re-frame each as an AEAD envelope bound to
            // its own file-relative ordinal, preserving per-record framing (so a
            // torn tail discards exactly one record).
            let buf = &buffer.buffer;
            let mut pos = 0usize;
            let mut total_written = 0u64;
            while pos + 4 <= buf.len() {
                let content_len =
                    u32::from_le_bytes([buf[pos], buf[pos + 1], buf[pos + 2], buf[pos + 3]])
                        as usize;
                pos += 4;
                if pos + content_len > buf.len() {
                    return Err(SochDBError::Corruption(
                        "txn buffer truncated mid-record during encrypted flush".into(),
                    ));
                }
                let body = &buf[pos..pos + content_len];
                pos += content_len;
                let ord = self.records_in_file.load(Ordering::SeqCst);
                let frame = self.encrypt_frame(body, ord)?;
                writer.write_all(&frame)?;
                // Advance per record only after its bytes are committed, so a
                // mid-batch write failure leaves records_in_file == records
                // actually written (no ordinal desync for the survivors).
                self.records_in_file.fetch_add(1, Ordering::SeqCst);
                total_written += frame.len() as u64;
            }
            let seq = self
                .sequence
                .fetch_add(buffer.entry_count as u64, Ordering::SeqCst);
            self.bytes_since_sync
                .fetch_add(total_written, Ordering::Relaxed);
            return Ok(seq);
        }

        writer.write_all(&buffer.buffer)?;

        let seq = self
            .sequence
            .fetch_add(buffer.entry_count as u64, Ordering::SeqCst);
        self.bytes_since_sync
            .fetch_add(buffer.buffer.len() as u64, Ordering::Relaxed);

        Ok(seq)
    }

    /// Get the current size of the WAL file in bytes
    pub fn size_bytes(&self) -> u64 {
        std::fs::metadata(&self.path).map(|m| m.len()).unwrap_or(0)
    }

    /// Allocate a new transaction ID
    pub fn alloc_txn_id(&self) -> u64 {
        self.next_txn_id.fetch_add(1, Ordering::SeqCst)
    }

    /// Begin a new transaction
    pub fn begin_transaction(&self) -> Result<u64> {
        let txn_id = self.alloc_txn_id();
        let entry = TxnWalEntry::txn_begin(txn_id);
        self.append(&entry)?;
        Ok(txn_id)
    }

    /// Commit a transaction (with fsync for durability)
    ///
    /// This flushes all pending writes and then fsyncs the commit record.
    ///
    /// # Durability Guarantee
    ///
    /// After this method returns `Ok(())`, the commit record is durable on disk.
    /// The transaction is guaranteed to survive process crash, OS crash, and
    /// power failure (assuming the storage device honors fsync correctly).
    ///
    /// # Performance
    ///
    /// Each call performs an fsync (~5ms on HDD, ~0.1ms on NVMe).
    /// For high-throughput workloads, use `EventDrivenGroupCommit` which
    /// batches multiple commits into a single fsync via `commit_durable_batch()`.
    pub fn commit_transaction(&self, txn_id: u64) -> Result<()> {
        // First flush any pending buffered writes
        self.flush()?;

        // Then write commit record with fsync
        let entry = TxnWalEntry::txn_commit(txn_id);
        self.append_sync(&entry)?;
        Ok(())
    }

    /// Commit a batch of transactions with a single fsync (group commit).
    ///
    /// Writes commit records for all transaction IDs, then performs a single
    /// flush + fsync. This amortizes the fsync cost across N transactions,
    /// achieving ~N× throughput improvement over individual commits.
    ///
    /// # Durability Guarantee
    ///
    /// After this method returns `Ok(())`, ALL transactions in the batch
    /// are durable on disk. Either all commit records are visible after
    /// crash recovery, or none are (atomic batch durability).
    ///
    /// # Usage
    ///
    /// This method is called by `EventDrivenGroupCommit::flush_fn` to
    /// implement the group commit pattern. Do not call directly unless
    /// you are implementing your own commit batching.
    pub fn commit_durable_batch(&self, txn_ids: &[u64]) -> Result<()> {
        // Write all commit records without flushing (batch them in BufWriter)
        for &txn_id in txn_ids {
            let entry = TxnWalEntry::txn_commit(txn_id);
            self.append_no_flush(&entry)?;
        }

        // Single flush + fsync for the entire batch
        self.flush()?;
        self.sync()?;
        Ok(())
    }

    /// Abort a transaction
    pub fn abort_transaction(&self, txn_id: u64) -> Result<()> {
        let entry = TxnWalEntry::txn_abort(txn_id);
        self.append(&entry)?;
        Ok(())
    }

    /// Write data within a transaction
    pub fn write(&self, txn_id: u64, key: Vec<u8>, value: Vec<u8>) -> Result<u64> {
        let entry = TxnWalEntry::data(txn_id, key, value);
        self.append(&entry)
    }

    /// Replay WAL for crash recovery
    ///
    /// Returns (committed_writes, recovered_txn_count)
    #[allow(clippy::type_complexity)]
    pub fn replay_for_recovery(&self) -> Result<(Vec<(Vec<u8>, Vec<u8>)>, usize)> {
        let file = File::open(&self.path)?;
        let mut reader = BufReader::new(file);

        let mut pending_writes: std::collections::HashMap<u64, Vec<(Vec<u8>, Vec<u8>)>> =
            std::collections::HashMap::new();
        let mut result = Vec::new();
        let mut txn_count = 0;
        let mut ordinal: u64 = 0;

        // Single pass in WAL order. Transaction ids are process-local and
        // short-lived CLI processes can reuse the same ids after restart, so
        // grouping the entire WAL by txn_id would merge unrelated transactions
        // and replay older writes after newer ones.
        loop {
            match self.read_record(&mut reader, &mut ordinal) {
                Ok(entry) => match entry.record_type {
                    WalRecordType::TxnBegin => {
                        pending_writes.insert(entry.txn_id, Vec::new());
                    }
                    WalRecordType::Data => {
                        // Accept data for any txn_id we've seen a Begin for,
                        // and also for txn_ids without a Begin (they might have
                        // their Begin later in the WAL due to buffered writes).
                        pending_writes
                            .entry(entry.txn_id)
                            .or_insert_with(Vec::new)
                            .push((entry.key, entry.value));
                    }
                    WalRecordType::TxnCommit => {
                        if let Some(writes) = pending_writes.remove(&entry.txn_id) {
                            result.extend(writes);
                            txn_count += 1;
                        }
                    }
                    WalRecordType::TxnAbort => {
                        pending_writes.remove(&entry.txn_id);
                    }
                    _ => {}
                },
                Err(SochDBError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    break;
                }
                Err(e) => {
                    // Encrypted WALs must fail loud on a non-EOF error so a wrong
                    // key / tamper / corruption can never masquerade as EOF and
                    // silently drop committed data. (Wrong key is already excluded
                    // earlier by the keyring canary, so this is genuine corruption
                    // or tampering.) Plaintext keeps the legacy torn-tail tolerance.
                    if self.encryption.is_enabled() {
                        return Err(e);
                    }
                    break;
                }
            }
        }

        Ok((result, txn_count))
    }

    /// Replay the WAL up to a PITR target, returning the committed writes whose
    /// transactions are included by the target (and the count of such txns).
    ///
    /// Prefix semantics — recovery "rolls forward" through the WAL and STOPS at
    /// the target, so the result is always a state the database actually passed
    /// through:
    /// - [`RecoveryTarget::Lsn`]`(l)`: include the first `l` WAL records (exact;
    ///   `l` matches [`Self::sequence`] / `DurableStorage::current_lsn` captured
    ///   at that point). A transaction whose commit lands after record `l` is
    ///   excluded (atomic — partial transactions are never applied).
    /// - [`RecoveryTarget::Timestamp`]`(t)`: stop at the FIRST transaction whose
    ///   commit timestamp exceeds `t`. Best-effort on the coarse, possibly
    ///   non-monotonic commit clock; prefer `Lsn` for an exact cut.
    ///
    /// Crypto-aware and fail-loud, exactly like [`Self::replay_for_recovery`].
    #[allow(clippy::type_complexity)]
    pub fn replay_to_target(
        &self,
        target: RecoveryTarget,
    ) -> Result<(Vec<(Vec<u8>, Vec<u8>)>, usize)> {
        let file = File::open(&self.path)?;
        let mut reader = BufReader::new(file);

        let mut pending_writes: std::collections::HashMap<u64, Vec<(Vec<u8>, Vec<u8>)>> =
            std::collections::HashMap::new();
        let mut result = Vec::new();
        let mut txn_count = 0;
        let mut ordinal: u64 = 0;

        loop {
            match self.read_record(&mut reader, &mut ordinal) {
                Ok(entry) => {
                    // LSN target: `ordinal` is now this record's 1-based LSN.
                    // Stop once we pass the target (this record is excluded).
                    if let RecoveryTarget::Lsn(l) = target {
                        if ordinal > l {
                            break;
                        }
                    }
                    match entry.record_type {
                        WalRecordType::TxnBegin => {
                            pending_writes.insert(entry.txn_id, Vec::new());
                        }
                        WalRecordType::Data => {
                            pending_writes
                                .entry(entry.txn_id)
                                .or_insert_with(Vec::new)
                                .push((entry.key, entry.value));
                        }
                        WalRecordType::TxnCommit => {
                            // Timestamp target: stop rolling forward at the first
                            // transaction committed AFTER the target (exclude it).
                            if let RecoveryTarget::Timestamp(t) = target {
                                if entry.timestamp_us > t {
                                    break;
                                }
                            }
                            if let Some(writes) = pending_writes.remove(&entry.txn_id) {
                                result.extend(writes);
                                txn_count += 1;
                            }
                        }
                        WalRecordType::TxnAbort => {
                            pending_writes.remove(&entry.txn_id);
                        }
                        _ => {}
                    }
                }
                Err(SochDBError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    break;
                }
                Err(e) => {
                    if self.encryption.is_enabled() {
                        return Err(e);
                    }
                    break;
                }
            }
        }

        Ok((result, txn_count))
    }

    /// Replay WAL with a callback
    pub fn replay<F>(&self, mut callback: F) -> Result<u64>
    where
        F: FnMut(TxnWalEntry) -> Result<()>,
    {
        let file = File::open(&self.path)?;
        let mut reader = BufReader::new(file);
        let mut count = 0u64;
        let mut ordinal: u64 = 0;

        loop {
            match self.read_record(&mut reader, &mut ordinal) {
                Ok(entry) => {
                    callback(entry)?;
                    count += 1;
                }
                Err(SochDBError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    break;
                }
                Err(e) => {
                    // Encrypted: fail loud (never silently drop committed data on a
                    // non-EOF error). Plaintext: tolerate a trailing incomplete entry.
                    if self.encryption.is_enabled() {
                        return Err(e);
                    }
                    eprintln!("WAL replay warning: {:?}", e);
                    break;
                }
            }
        }

        Ok(count)
    }

    /// Truncate WAL (called after successful checkpoint)
    ///
    /// Flushes any buffered writes, truncates the file to 0 bytes,
    /// and resets sequence counters. The file is opened in `O_APPEND`
    /// mode so subsequent writes will correctly start at offset 0.
    ///
    /// **WARNING**: After truncation, all data durability is lost.
    /// The in-memory memtable still holds data for the current session,
    /// but a crash after truncation means the data cannot be recovered
    /// from the WAL.
    pub fn truncate(&self) -> Result<()> {
        let mut writer = self.writer.lock();
        // Flush BufWriter so no stale data is written after truncation
        writer.flush()?;
        let file = writer.get_ref();
        file.set_len(0)?;
        file.sync_all()?;
        self.sequence.store(0, Ordering::SeqCst);
        self.bytes_since_sync.store(0, Ordering::Relaxed);
        // The file restarts at offset 0, so per-record AAD ordinals restart too;
        // a fresh reader of the truncated file will count from 0 to match.
        self.records_in_file.store(0, Ordering::SeqCst);
        Ok(())
    }

    /// Write a checkpoint marker
    pub fn write_checkpoint(&self) -> Result<u64> {
        let entry = TxnWalEntry::checkpoint(0);
        self.append_sync(&entry)
    }

    /// Write a Compensation Log Record (CLR) for undo operations
    ///
    /// CLRs are redo-only records that skip past already undone operations
    /// during recovery. The undo_next_lsn field tells recovery where to
    /// continue undoing after this CLR.
    pub fn append_clr(
        &self,
        txn_id: u64,
        _original_lsn: u64,
        undo_next_lsn: Option<u64>,
        undo_data: &[u8],
    ) -> Result<u64> {
        // Encode undo_next_lsn into the key field
        let key = undo_next_lsn.unwrap_or(0).to_le_bytes().to_vec();
        let entry = TxnWalEntry {
            record_type: WalRecordType::CompensationLogRecord,
            txn_id,
            timestamp_us: TxnWalEntry::now_us(),
            key, // undo_next_lsn encoded in key
            value: undo_data.to_vec(),
        };
        self.append(&entry)
    }

    /// Write checkpoint with data (for fuzzy checkpoints)
    pub fn write_checkpoint_with_data(&self, checkpoint_data: &[u8]) -> Result<u64> {
        let entry = TxnWalEntry {
            record_type: WalRecordType::Checkpoint,
            txn_id: 0,
            timestamp_us: TxnWalEntry::now_us(),
            key: Vec::new(),
            value: checkpoint_data.to_vec(),
        };
        self.append_sync(&entry)
    }

    /// Write checkpoint end with captured state
    pub fn write_checkpoint_end(&self, checkpoint_data: &[u8]) -> Result<u64> {
        let entry = TxnWalEntry {
            record_type: WalRecordType::CheckpointEnd,
            txn_id: 0,
            timestamp_us: TxnWalEntry::now_us(),
            key: Vec::new(),
            value: checkpoint_data.to_vec(),
        };
        self.append_sync(&entry)
    }

    /// Get current sequence number
    pub fn sequence(&self) -> u64 {
        self.sequence.load(Ordering::SeqCst)
    }

    /// Get bytes written since last sync
    pub fn bytes_since_sync(&self) -> u64 {
        self.bytes_since_sync.load(Ordering::Relaxed)
    }

    /// Get path to WAL file
    pub fn path(&self) -> &Path {
        &self.path
    }
}

/// Statistics about WAL state
#[derive(Debug, Clone, Default)]
pub struct TxnWalStats {
    /// Number of entries written
    pub entries_written: u64,
    /// Bytes written since last sync
    pub bytes_since_sync: u64,
    /// Current transaction ID counter
    pub next_txn_id: u64,
}

// ============================================================================
// Sharded WAL for Reduced Mutex Contention
// ============================================================================

/// Sharded Write-Ahead Log for high-concurrency workloads
///
/// Instead of a single Mutex<File>, uses multiple shards to reduce contention:
/// - Writers hash to shard by txn_id
/// - Each shard has its own buffer
/// - Central coordinator handles fsync ordering
///
/// Reduces contention from O(1) bottleneck to O(num_shards) parallelism.
#[allow(dead_code)]
pub struct ShardedWal {
    /// Shard writers (txn_id % num_shards selects shard)
    shards: Vec<parking_lot::Mutex<WalShard>>,
    /// Number of shards (power of 2)
    num_shards: usize,
    /// Central WAL file for ordered writes
    central_writer: parking_lot::Mutex<BufWriter<File>>,
    /// Next transaction ID
    next_txn_id: AtomicU64,
    /// Write sequence (global ordering)
    sequence: AtomicU64,
    /// Path
    path: PathBuf,
}

/// Individual WAL shard buffer
struct WalShard {
    /// Buffered entries for this shard
    buffer: Vec<u8>,
    /// Number of entries buffered
    entry_count: usize,
}

impl WalShard {
    fn new() -> Self {
        Self {
            buffer: Vec::with_capacity(64 * 1024), // 64KB per shard
            entry_count: 0,
        }
    }

    fn append(&mut self, entry: &TxnWalEntry) {
        let bytes = entry.to_bytes();
        self.buffer.extend_from_slice(&bytes);
        self.entry_count += 1;
    }

    fn is_empty(&self) -> bool {
        self.buffer.is_empty()
    }

    fn drain(&mut self) -> Vec<u8> {
        self.entry_count = 0;
        std::mem::take(&mut self.buffer)
    }
}

impl ShardedWal {
    /// Create sharded WAL with specified number of shards
    ///
    /// Recommended: 4-8 shards for typical server workloads
    pub fn new<P: AsRef<Path>>(path: P, num_shards: usize) -> Result<Self> {
        let path = path.as_ref().to_path_buf();

        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        let file = std::fs::OpenOptions::new()
            .create(true)
            .append(true)
            .read(true)
            .open(&path)?;

        // Round up to power of 2 for fast modulo
        let num_shards = num_shards.next_power_of_two();
        let shards: Vec<_> = (0..num_shards)
            .map(|_| parking_lot::Mutex::new(WalShard::new()))
            .collect();

        Ok(Self {
            shards,
            num_shards,
            central_writer: parking_lot::Mutex::new(BufWriter::with_capacity(256 * 1024, file)),
            next_txn_id: AtomicU64::new(1),
            sequence: AtomicU64::new(0),
            path,
        })
    }

    /// Get shard index for transaction
    #[inline]
    fn shard_idx(&self, txn_id: u64) -> usize {
        (txn_id as usize) & (self.num_shards - 1)
    }

    /// Append entry to appropriate shard (lock-free for different txns)
    pub fn append(&self, entry: &TxnWalEntry) -> u64 {
        let shard_idx = self.shard_idx(entry.txn_id);
        let mut shard = self.shards[shard_idx].lock();
        shard.append(entry);
        self.sequence.fetch_add(1, Ordering::SeqCst)
    }

    /// Allocate transaction ID
    pub fn alloc_txn_id(&self) -> u64 {
        self.next_txn_id.fetch_add(1, Ordering::SeqCst)
    }

    /// Flush all shard buffers to central file
    pub fn flush(&self) -> Result<()> {
        let mut central = self.central_writer.lock();

        // Collect all shard buffers (brief lock per shard)
        for shard in &self.shards {
            let mut shard_guard = shard.lock();
            if !shard_guard.is_empty() {
                let data = shard_guard.drain();
                central.write_all(&data)?;
            }
        }

        central.flush()?;
        Ok(())
    }

    /// Sync to disk (fsync)
    pub fn sync(&self) -> Result<()> {
        self.flush()?;
        let central = self.central_writer.lock();
        central.get_ref().sync_all()?;
        Ok(())
    }

    /// Begin transaction
    pub fn begin_transaction(&self) -> Result<u64> {
        let txn_id = self.alloc_txn_id();
        let entry = TxnWalEntry::txn_begin(txn_id);
        self.append(&entry);
        Ok(txn_id)
    }

    /// Write data
    pub fn write(&self, txn_id: u64, key: Vec<u8>, value: Vec<u8>) -> Result<u64> {
        let entry = TxnWalEntry::data(txn_id, key, value);
        Ok(self.append(&entry))
    }

    /// Commit transaction
    pub fn commit_transaction(&self, txn_id: u64) -> Result<u64> {
        let entry = TxnWalEntry::txn_commit(txn_id);
        let seq = self.append(&entry);
        self.sync()?; // Fsync on commit for durability
        Ok(seq)
    }

    /// Get statistics
    pub fn stats(&self) -> ShardedWalStats {
        let mut shard_entry_counts = Vec::with_capacity(self.num_shards);
        for shard in &self.shards {
            shard_entry_counts.push(shard.lock().entry_count);
        }

        ShardedWalStats {
            num_shards: self.num_shards,
            total_entries: self.sequence.load(Ordering::SeqCst),
            next_txn_id: self.next_txn_id.load(Ordering::SeqCst),
            shard_entry_counts,
        }
    }
}

/// Statistics for sharded WAL
#[derive(Debug, Clone)]
pub struct ShardedWalStats {
    pub num_shards: usize,
    pub total_entries: u64,
    pub next_txn_id: u64,
    pub shard_entry_counts: Vec<usize>,
}

/// Detailed crash recovery statistics
#[derive(Debug, Clone, Default)]
pub struct CrashRecoveryStats {
    /// Total records read from WAL
    pub total_records: u64,
    /// Number of committed transactions
    pub committed_txns: u64,
    /// Number of uncommitted (rolled back) transactions
    pub rolled_back_txns: u64,
    /// Number of explicitly aborted transactions
    pub aborted_txns: u64,
    /// Number of data writes recovered
    pub recovered_writes: u64,
    /// Number of torn/corrupted records at end (expected on crash)
    pub torn_records: u64,
    /// Bytes read from WAL
    pub bytes_read: u64,
    /// Recovery duration in microseconds
    pub recovery_duration_us: u64,
    /// Highest transaction ID seen (for restarting counter)
    pub max_txn_id: u64,
}

impl TxnWal {
    /// Get WAL statistics
    pub fn stats(&self) -> TxnWalStats {
        TxnWalStats {
            entries_written: self.sequence.load(Ordering::SeqCst),
            bytes_since_sync: self.bytes_since_sync.load(Ordering::Relaxed),
            next_txn_id: self.next_txn_id.load(Ordering::SeqCst),
        }
    }

    /// Full crash recovery with detailed statistics
    ///
    /// This method provides ACID recovery guarantees:
    /// 1. **Atomicity**: Uncommitted transactions are rolled back
    /// 2. **Durability**: All committed transactions are replayed
    /// 3. **Torn Write Detection**: Partial records at EOF are detected via CRC32
    ///
    /// Returns (committed_writes, stats) where committed_writes contains
    /// all key-value pairs from committed transactions in order.
    #[allow(clippy::type_complexity)]
    pub fn crash_recovery(&self) -> Result<(Vec<(Vec<u8>, Vec<u8>)>, CrashRecoveryStats)> {
        let start_time = std::time::Instant::now();
        let file = File::open(&self.path)?;
        let file_size = file.metadata()?.len();
        let mut reader = BufReader::new(file);

        let mut stats = CrashRecoveryStats {
            bytes_read: file_size,
            ..Default::default()
        };

        let mut committed_txns: HashSet<u64> = HashSet::new();
        let mut aborted_txns: HashSet<u64> = HashSet::new();
        let mut pending_writes: std::collections::HashMap<u64, Vec<(Vec<u8>, Vec<u8>)>> =
            std::collections::HashMap::new();
        let mut all_txns: HashSet<u64> = HashSet::new();

        // Read all records, stopping at first corruption (torn write)
        let mut ordinal: u64 = 0;
        loop {
            match self.read_record(&mut reader, &mut ordinal) {
                Ok(entry) => {
                    stats.total_records += 1;
                    if entry.txn_id > stats.max_txn_id {
                        stats.max_txn_id = entry.txn_id;
                    }

                    match entry.record_type {
                        WalRecordType::TxnBegin => {
                            pending_writes.insert(entry.txn_id, Vec::new());
                            all_txns.insert(entry.txn_id);
                        }
                        WalRecordType::Data => {
                            if let Some(writes) = pending_writes.get_mut(&entry.txn_id) {
                                writes.push((entry.key, entry.value));
                            }
                        }
                        WalRecordType::TxnCommit => {
                            committed_txns.insert(entry.txn_id);
                        }
                        WalRecordType::TxnAbort => {
                            pending_writes.remove(&entry.txn_id);
                            aborted_txns.insert(entry.txn_id);
                        }
                        _ => {}
                    }
                }
                Err(SochDBError::Io(e)) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    // Clean EOF
                    break;
                }
                Err(e) => {
                    // Encrypted: an AEAD auth failure / tamper / corruption is a
                    // hard error — abort recovery rather than silently truncating
                    // committed data (wrong key is already caught by the keyring
                    // canary at open). Plaintext: treat trailing corruption as a
                    // torn write and stop.
                    if self.encryption.is_enabled() {
                        return Err(e);
                    }
                    stats.torn_records += 1;
                    break;
                }
            }
        }

        // Collect committed writes
        let mut result = Vec::new();
        for (txn_id, writes) in &pending_writes {
            if committed_txns.contains(txn_id) {
                stats.committed_txns += 1;
                stats.recovered_writes += writes.len() as u64;
                result.extend(writes.clone());
            }
        }

        // Count aborted and rolled-back transactions
        stats.aborted_txns = aborted_txns.len() as u64;
        stats.rolled_back_txns = all_txns.len() as u64 - stats.committed_txns - stats.aborted_txns;

        stats.recovery_duration_us = start_time.elapsed().as_micros() as u64;

        Ok((result, stats))
    }
}

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

    #[test]
    fn test_wal_entry_roundtrip() {
        let entry = TxnWalEntry::data(42, b"key".to_vec(), b"value".to_vec());
        let bytes = entry.to_bytes();

        let mut cursor = std::io::Cursor::new(bytes);
        let recovered = TxnWalEntry::from_reader(&mut cursor).unwrap();

        assert_eq!(recovered.record_type, WalRecordType::Data);
        assert_eq!(recovered.txn_id, 42);
        assert_eq!(recovered.key, b"key");
        assert_eq!(recovered.value, b"value");
    }

    #[test]
    fn test_wal_append_and_replay() {
        let dir = tempdir().unwrap();
        let wal_path = dir.path().join("test.wal");

        // Write some entries
        {
            let wal = TxnWal::new(&wal_path).unwrap();
            let txn_id = wal.begin_transaction().unwrap();
            wal.write(txn_id, b"k1".to_vec(), b"v1".to_vec()).unwrap();
            wal.write(txn_id, b"k2".to_vec(), b"v2".to_vec()).unwrap();
            wal.commit_transaction(txn_id).unwrap();
        }

        // Replay and verify
        {
            let wal = TxnWal::new(&wal_path).unwrap();
            let (writes, txn_count) = wal.replay_for_recovery().unwrap();

            assert_eq!(txn_count, 1);
            assert_eq!(writes.len(), 2);
            assert_eq!(writes[0], (b"k1".to_vec(), b"v1".to_vec()));
            assert_eq!(writes[1], (b"k2".to_vec(), b"v2".to_vec()));
        }
    }

    #[test]
    fn test_uncommitted_transaction_rollback() {
        let dir = tempdir().unwrap();
        let wal_path = dir.path().join("test.wal");

        // Write committed and uncommitted transactions
        {
            let wal = TxnWal::new(&wal_path).unwrap();

            // Committed transaction
            let txn1 = wal.begin_transaction().unwrap();
            wal.write(txn1, b"committed".to_vec(), b"yes".to_vec())
                .unwrap();
            wal.commit_transaction(txn1).unwrap();

            // Uncommitted transaction (simulates crash)
            let txn2 = wal.begin_transaction().unwrap();
            wal.write(txn2, b"uncommitted".to_vec(), b"no".to_vec())
                .unwrap();
            // No commit!
        }

        // Replay - uncommitted should be rolled back
        {
            let wal = TxnWal::new(&wal_path).unwrap();
            let (writes, txn_count) = wal.replay_for_recovery().unwrap();

            assert_eq!(txn_count, 1); // Only committed transaction
            assert_eq!(writes.len(), 1);
            assert_eq!(writes[0], (b"committed".to_vec(), b"yes".to_vec()));
        }
    }

    #[test]
    fn test_aborted_transaction() {
        let dir = tempdir().unwrap();
        let wal_path = dir.path().join("test.wal");

        {
            let wal = TxnWal::new(&wal_path).unwrap();

            let txn = wal.begin_transaction().unwrap();
            wal.write(txn, b"aborted".to_vec(), b"data".to_vec())
                .unwrap();
            wal.abort_transaction(txn).unwrap();
        }

        {
            let wal = TxnWal::new(&wal_path).unwrap();
            let (writes, txn_count) = wal.replay_for_recovery().unwrap();

            assert_eq!(txn_count, 0);
            assert!(writes.is_empty());
        }
    }

    #[test]
    fn test_checksum_validation() {
        let entry = TxnWalEntry::data(1, b"key".to_vec(), b"value".to_vec());
        let mut bytes = entry.to_bytes();

        // Corrupt the checksum
        let len = bytes.len();
        bytes[len - 1] ^= 0xFF;

        let mut cursor = std::io::Cursor::new(bytes);
        let result = TxnWalEntry::from_reader(&mut cursor);

        assert!(result.is_err());
    }

    #[test]
    fn test_crash_recovery_with_stats() {
        let dir = tempdir().unwrap();
        let wal_path = dir.path().join("test.wal");

        // Simulate complex workload
        {
            let wal = TxnWal::new(&wal_path).unwrap();

            // Committed transaction 1
            let txn1 = wal.begin_transaction().unwrap();
            wal.write(txn1, b"k1".to_vec(), b"v1".to_vec()).unwrap();
            wal.write(txn1, b"k2".to_vec(), b"v2".to_vec()).unwrap();
            wal.commit_transaction(txn1).unwrap();

            // Aborted transaction
            let txn2 = wal.begin_transaction().unwrap();
            wal.write(txn2, b"aborted_key".to_vec(), b"aborted_val".to_vec())
                .unwrap();
            wal.abort_transaction(txn2).unwrap();

            // Committed transaction 2
            let txn3 = wal.begin_transaction().unwrap();
            wal.write(txn3, b"k3".to_vec(), b"v3".to_vec()).unwrap();
            wal.commit_transaction(txn3).unwrap();

            // Uncommitted transaction (simulates crash)
            let txn4 = wal.begin_transaction().unwrap();
            wal.write(txn4, b"uncommitted".to_vec(), b"data".to_vec())
                .unwrap();
            // No commit - simulates crash
        }

        // Recover and verify
        {
            let wal = TxnWal::new(&wal_path).unwrap();
            let (writes, stats) = wal.crash_recovery().unwrap();

            // Should have 3 writes from 2 committed transactions
            assert_eq!(writes.len(), 3);
            assert_eq!(stats.committed_txns, 2);
            assert_eq!(stats.aborted_txns, 1);
            assert_eq!(stats.rolled_back_txns, 1); // txn4 was uncommitted
            assert_eq!(stats.recovered_writes, 3);
            assert!(stats.recovery_duration_us > 0);
        }
    }

    #[test]
    fn test_torn_write_detection() {
        let dir = tempdir().unwrap();
        let wal_path = dir.path().join("test.wal");

        // Write a valid transaction
        {
            let wal = TxnWal::new(&wal_path).unwrap();
            let txn = wal.begin_transaction().unwrap();
            wal.write(txn, b"key".to_vec(), b"value".to_vec()).unwrap();
            wal.commit_transaction(txn).unwrap();
        }

        // Append corrupted bytes to simulate torn write
        {
            use std::io::Write;
            let mut file = std::fs::OpenOptions::new()
                .append(true)
                .open(&wal_path)
                .unwrap();
            // Write partial record (torn write)
            file.write_all(&[0x10, 0x00, 0x00, 0x00, 0xFF, 0xFF])
                .unwrap();
        }

        // Recovery should still work, detecting torn write
        {
            let wal = TxnWal::new(&wal_path).unwrap();
            let (writes, stats) = wal.crash_recovery().unwrap();

            // Should recover the valid transaction
            assert_eq!(writes.len(), 1);
            assert_eq!(stats.committed_txns, 1);
            assert_eq!(stats.torn_records, 1);
        }
    }

    #[test]
    fn test_crc32_determinism() {
        // Verify CRC32 produces consistent checksums for same content
        let mut entry1 = TxnWalEntry::data(42, b"key".to_vec(), b"value".to_vec());
        entry1.timestamp_us = 12345; // Fixed timestamp for determinism

        let mut entry2 = TxnWalEntry::data(42, b"key".to_vec(), b"value".to_vec());
        entry2.timestamp_us = 12345; // Same timestamp

        assert_eq!(entry1.checksum(), entry2.checksum());

        // Different content should produce different checksum
        let mut entry3 = TxnWalEntry::data(42, b"key".to_vec(), b"different".to_vec());
        entry3.timestamp_us = 12345;
        assert_ne!(entry1.checksum(), entry3.checksum());

        // Verify roundtrip preserves checksum
        let bytes = entry1.to_bytes();
        let mut cursor = std::io::Cursor::new(bytes);
        let recovered = TxnWalEntry::from_reader(&mut cursor).unwrap();
        assert_eq!(recovered.checksum(), entry1.checksum());
    }
}

#[cfg(test)]
mod encryption_wal_tests {
    use super::*;
    use crate::encryption::EncryptionEngine;
    use tempfile::tempdir;

    fn enc(key: u8) -> Arc<EncryptionEngine> {
        Arc::new(EncryptionEngine::new(&[key; 32]))
    }
    const UUID: [u8; 16] = [9u8; 16];

    /// Write a committed txn (begin/data/commit) and one aborted txn via the
    /// encrypted paths, then reopen with the same key and crash-recover.
    #[test]
    fn encrypted_write_then_recover_roundtrip() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("enc.wal");

        {
            let wal = TxnWal::new_with_encryption(&path, enc(7), UUID, 0).unwrap();
            // committed txn via append + write_no_flush_refs + flush_buffer paths
            let t1 = wal.begin_transaction().unwrap();
            wal.write(t1, b"alpha".to_vec(), b"one".to_vec()).unwrap();
            wal.write_no_flush_refs(t1, b"beta", b"two").unwrap();
            // also exercise the buffer/flush_buffer batch path
            let mut buf = TxnWalBuffer::new(t1);
            buf.append(b"gamma", b"three");
            wal.flush_buffer(&buf).unwrap();
            wal.commit_transaction(t1).unwrap();

            // aborted txn must NOT survive
            let t2 = wal.begin_transaction().unwrap();
            wal.write(t2, b"ghost".to_vec(), b"x".to_vec()).unwrap();
            wal.abort_transaction(t2).unwrap();
            wal.sync().unwrap();
        }

        // On-disk bytes must NOT contain the plaintext values.
        let raw = std::fs::read(&path).unwrap();
        assert!(!contains(&raw, b"alpha"));
        assert!(!contains(&raw, b"three"));

        let wal = TxnWal::new_with_encryption(&path, enc(7), UUID, 0).unwrap();
        let (writes, stats) = wal.crash_recovery().unwrap();
        let keys: Vec<_> = writes.iter().map(|(k, _)| k.clone()).collect();
        assert!(keys.contains(&b"alpha".to_vec()));
        assert!(keys.contains(&b"beta".to_vec()));
        assert!(keys.contains(&b"gamma".to_vec()));
        assert!(!keys.contains(&b"ghost".to_vec()), "aborted txn leaked");
        assert_eq!(stats.committed_txns, 1);
    }

    /// Wrong key at the WAL layer must FAIL LOUD, not silently return empty.
    /// (In production the keyring canary catches this even earlier; this proves
    /// the replay path itself never swallows an AEAD failure as EOF.)
    #[test]
    fn wrong_key_fails_loud_not_empty() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("enc.wal");
        {
            let wal = TxnWal::new_with_encryption(&path, enc(1), UUID, 0).unwrap();
            let t = wal.begin_transaction().unwrap();
            wal.write(t, b"k".to_vec(), b"v".to_vec()).unwrap();
            wal.commit_transaction(t).unwrap();
            wal.sync().unwrap();
        }
        // Reopen with the WRONG key: recover_state runs in the constructor and
        // must surface a hard error rather than an "empty" success.
        let opened = TxnWal::new_with_encryption(&path, enc(2), UUID, 0);
        assert!(opened.is_err(), "wrong key opened silently (data-loss bug)");
        match opened.err().unwrap() {
            SochDBError::Encryption(_) => {}
            other => panic!("expected Encryption error, got {other:?}"),
        }
    }

    /// Tampering with a committed encrypted record must fail recovery loud.
    #[test]
    fn tamper_midstream_fails_loud() {
        let dir = tempdir().unwrap();
        let path = dir.path().join("enc.wal");
        {
            let wal = TxnWal::new_with_encryption(&path, enc(5), UUID, 0).unwrap();
            let t = wal.begin_transaction().unwrap();
            wal.write(t, b"k1".to_vec(), b"v1".to_vec()).unwrap();
            wal.commit_transaction(t).unwrap();
            wal.sync().unwrap();
        }
        // Flip a byte well inside the file (not the final torn-tail region).
        let mut raw = std::fs::read(&path).unwrap();
        let mid = raw.len() / 2;
        raw[mid] ^= 0xFF;
        std::fs::write(&path, &raw).unwrap();

        let opened = TxnWal::new_with_encryption(&path, enc(5), UUID, 0);
        // Either the constructor's recover_state errors, or a later crash_recovery
        // does — but it must NEVER silently accept tampered ciphertext.
        let failed = opened.is_err()
            || opened
                .ok()
                .map(|w| w.crash_recovery().is_err())
                .unwrap_or(true);
        assert!(failed, "tampered encrypted WAL was silently accepted");
    }

    /// The disabled engine must write a record's bytes EXACTLY as the legacy
    /// `to_bytes()` frame (no added encryption framing / format drift), while the
    /// enabled engine must NOT (it adds the AEAD envelope and is unreadable by the
    /// plaintext reader). Records embed wall-clock timestamps, so we compare a
    /// single fixed entry object against its own `to_bytes()` for determinism.
    #[test]
    fn disabled_engine_is_byte_identical_to_plaintext() {
        let dir = tempdir().unwrap();

        let entry = TxnWalEntry::data(42, b"k1".to_vec(), b"v1".to_vec());
        let golden = entry.to_bytes();

        // Disabled engine: file bytes == to_bytes() exactly.
        let p_dis = dir.path().join("disabled.wal");
        {
            let wal = TxnWal::new_with_encryption(
                &p_dis,
                Arc::new(EncryptionEngine::disabled()),
                [0u8; 16],
                0,
            )
            .unwrap();
            wal.append(&entry).unwrap();
            wal.sync().unwrap();
        }
        assert_eq!(
            std::fs::read(&p_dis).unwrap(),
            golden,
            "disabled-engine append diverged from legacy plaintext frame"
        );

        // Enabled engine: file bytes differ and are NOT plaintext-parseable.
        let p_enc = dir.path().join("enc.wal");
        {
            let wal = TxnWal::new_with_encryption(&p_enc, enc(3), UUID, 0).unwrap();
            wal.append(&entry).unwrap();
            wal.sync().unwrap();
        }
        let enc_bytes = std::fs::read(&p_enc).unwrap();
        assert_ne!(enc_bytes, golden);
        let mut cur = std::io::Cursor::new(&enc_bytes);
        assert!(
            TxnWalEntry::from_reader(&mut cur).is_err()
                || cur.position() as usize != enc_bytes.len(),
            "ciphertext frame must not parse cleanly as a plaintext record"
        );
    }

    fn contains(haystack: &[u8], needle: &[u8]) -> bool {
        haystack.windows(needle.len()).any(|w| w == needle)
    }
}