hypercall/journal/

engine_journal_batcher.rs

//! WAL-first journal batcher for durable ACKs in the engine hot path.
//!
//! The engine pushes `JournalEntry` values into a bounded channel.
//! This task drains entries, appends them to a local WAL, calls `fdatasync`,
//! and only then allows ACK/event emission via `commit_ack`.
//!
//! PostgreSQL replication runs asynchronously after WAL durability so
//! client-visible latency is decoupled from network/database I/O.

use crate::observability::command_trace::EngineStateDigest;
#[cfg(feature = "rsm-state")]
use crate::rsm::commitment_store::{
    ValidatorRsmCommitmentStoreConfig, ValidatorRsmStateCommitmentRuntime,
};
#[cfg(feature = "rsm-state")]
use crate::rsm::engine_snapshot::EngineStateDigest as RsmEngineStateDigest;
use hypercall_journal::checkpoint::{
    checkpoint_path_for as checkpoint_path_for_wal, read_checkpoint, write_checkpoint,
    WalCheckpointMetadata,
};
#[cfg(feature = "rsm-state")]
mod rsm_block_persistence;
#[cfg(feature = "rsm-state")]
use hypercall_blocks::SigningAuthority;
#[cfg(feature = "rsm-state")]
use hypercall_blocks::{BlockLogEntry, BlockLogHeader};
use hypercall_db::EngineJournalBatchWriter;
#[cfg(feature = "rsm-state")]
use hypercall_db::ValidatorRsmStateReader;
use hypercall_journal::frame::{write_frame, WAL_CRC};
#[cfg(feature = "rsm-state")]
use hypercall_state_commitment::{
    leaves::GlobalLeaf,
    pipeline::{PreparedBatchCommitment, StateDelta},
};
use metrics::{counter, gauge, histogram};
#[cfg(feature = "rsm-state")]
use rsm_block_persistence::{RsmBlockPersistenceInput, RsmBlockPersistenceMode};
use rust_decimal::Decimal;
use serde::{Deserialize, Serialize};
use std::fs::{create_dir_all, File, OpenOptions};
use std::io::{BufWriter, Read, Seek, SeekFrom, Write};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, error, info, info_span, warn, Instrument};

#[cfg(target_os = "linux")]
use std::os::unix::io::AsRawFd;

/// Default channel capacity
pub const DEFAULT_CHANNEL_CAPACITY: usize = 3000;

const REPLICATION_RETRY_DELAY: Duration = Duration::from_millis(250);

/// Pre-allocate WAL file in 64 MiB chunks to avoid extent allocation stalls.
#[cfg(target_os = "linux")]
const WAL_PREALLOC_CHUNK: u64 = 64 * 1024 * 1024;

/// Messages sent from the batcher task to the dedicated WAL writer thread.
enum WalWriterMessage {
    WriteBatch {
        entries: Vec<JournalEntry>,
        ack_senders: Vec<Option<oneshot::Sender<()>>>,
        /// If Some, fired after fsync completes (used by Flush callers).
        completion: Option<oneshot::Sender<()>>,
    },
    /// Barrier: fires completion after all prior WriteBatches have been fsynced.
    /// Used by Flush when the batcher's local batch is empty but prior writes
    /// may still be in-flight on this thread.
    FlushBarrier {
        completion: oneshot::Sender<()>,
    },
    Shutdown,
}

/// Pre-serialized event payload ready for DB storage.
/// Contains wire-format bytes (version byte + msgpack) so no further serialization is needed.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct EventPayload {
    pub event_topic: String,
    pub event_key: Option<String>,
    pub event_data: Vec<u8>, // version byte + inner msgpack
    pub l2_sequence: Option<i64>,
    pub event_type_enum: hypercall_db_diesel::engine_enums::EventType,
}

/// Journal-only accounting deltas for a single OrderFilled event.
///
/// These are computed on the hot path from the pre-fill portfolio state and
/// persisted in WAL so the async journal replicator can materialize durable
/// ledger effects without re-running business logic against current state.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JournalFillSideEffect {
    pub event_idx: i32,
    pub trade_id: u64,
    pub taker_ledger_delta: Decimal,
    pub maker_ledger_delta: Decimal,
    pub taker_premium_delta: Decimal,
    pub maker_premium_delta: Decimal,
    #[serde(default)]
    pub underlying_notional: Option<Decimal>,
}

impl JournalFillSideEffect {
    pub fn from_fill_accounting(
        event_idx: i32,
        accounting: &hypercall_types::FillAccounting,
    ) -> Self {
        accounting.assert_cash_decomposition();
        Self {
            event_idx,
            trade_id: accounting.trade_id,
            taker_ledger_delta: accounting.taker_ledger_residual_delta(),
            maker_ledger_delta: accounting.maker_ledger_residual_delta(),
            taker_premium_delta: accounting.taker_premium_delta(),
            maker_premium_delta: accounting.maker_premium_delta(),
            underlying_notional: None,
        }
    }
}

/// Journal-only side effect for a CashWithdrawalUpdate command.
///
/// Persisted in WAL so the async journal replicator can materialize durable
/// DB balance decrements without re-running business logic.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct JournalCashWithdrawalSideEffect {
    pub wallet: hypercall_types::WalletAddress,
    pub request_id: String,
    pub amount: Decimal,
    pub balance_after: Decimal,
    pub timestamp_ms: u64,
}

/// A journal entry to be batched and durably appended.
#[derive(Debug)]
pub struct JournalEntry {
    pub received_ts_ms: u64,
    /// Wire-format bytes: [version: u8][msgpack payload]
    pub command_data: Vec<u8>,
    /// Wire-format bytes: [version: u8][msgpack payload]
    pub response_data: Option<Vec<u8>>,
    /// Extracted order_id for indexed column
    pub order_id: Option<i64>,
    pub pre_digest: EngineStateDigest,
    pub post_digest: EngineStateDigest,
    pub duration_ms: u64,
    pub events: Vec<EventPayload>,
    pub outbox_appends: Vec<crate::directive_outbox::DirectiveOutboxAppend>,
    pub fill_side_effects: Vec<JournalFillSideEffect>,
    pub cash_withdrawal_side_effect: Option<JournalCashWithdrawalSideEffect>,
    pub balance_updates: Vec<hypercall_types::BalanceUpdate>,
    pub created_at: Instant,
    /// Signalled after the entry is durably appended to WAL + fdatasync.
    pub commit_ack: Option<oneshot::Sender<()>>,
    pub request_uuid: hypercall_db_diesel::engine_enums::DbUuid,
    /// None for non-engine commands (e.g., TransactionRequests passthrough).
    pub command_type_enum: Option<hypercall_db_diesel::engine_enums::CommandType>,
    /// Command identity hash for validator RSM batch metadata.
    #[cfg(feature = "rsm-state")]
    pub command_identity_hash: [u8; 32],
    /// Full restart-sensitive engine state digest for RSM block roots.
    #[cfg(feature = "rsm-state")]
    pub rsm_state_digest: Option<RsmEngineStateDigest>,
}

/// Configuration for the journal batcher
#[derive(Debug, Clone)]
pub struct JournalBatcherConfig {
    /// Channel capacity (backpressure threshold)
    pub channel_capacity: usize,
    /// Maximum batch size for WAL append and replication chunks
    pub max_batch_size: usize,
    /// Maximum wait time before flushing a partial batch (ms)
    pub flush_interval_ms: u64,
    /// Whether to persist digests
    pub persist_digests: bool,
    /// WAL path used for durable local staging.
    pub wal_path: PathBuf,
    /// Environment partition for validator RSM block metadata.
    #[cfg(feature = "rsm-state")]
    pub rsm_environment: hypercall_db::ValidatorRsmEnvironment,
    #[cfg(feature = "rsm-state")]
    pub rsm_commitment_store: ValidatorRsmCommitmentStoreConfig,
}

impl Default for JournalBatcherConfig {
    fn default() -> Self {
        Self {
            channel_capacity: DEFAULT_CHANNEL_CAPACITY,
            max_batch_size: 100,
            flush_interval_ms: 10,
            persist_digests: true,
            wal_path: PathBuf::from(hypercall_journal::checkpoint::DEFAULT_WAL_PATH),
            #[cfg(feature = "rsm-state")]
            rsm_environment: hypercall_db::ValidatorRsmEnvironment::Development,
            #[cfg(feature = "rsm-state")]
            rsm_commitment_store: ValidatorRsmCommitmentStoreConfig {
                root_dir: PathBuf::from("target/validator-rsm-state"),
                prune_window: 10_000,
                pruning_enabled: false,
            },
        }
    }
}

impl JournalBatcherConfig {
    pub fn from_config(config: &crate::backend_config::JournalRuntimeConfig) -> Self {
        Self {
            channel_capacity: config.batch_channel_capacity,
            max_batch_size: config.batch_size,
            flush_interval_ms: config.flush_interval_ms,
            persist_digests: config.digests_enabled,
            wal_path: hypercall_journal::checkpoint::wal_path_from_config(config.wal_path.as_ref()),
            #[cfg(feature = "rsm-state")]
            rsm_environment: config.rsm_environment.into(),
            #[cfg(feature = "rsm-state")]
            rsm_commitment_store: ValidatorRsmCommitmentStoreConfig {
                root_dir: config.rsm_state.state_store_path.clone(),
                prune_window: config.rsm_state.prune_window_versions,
                pruning_enabled: config.rsm_state.pruning_enabled,
            },
        }
    }
}

/// Messages sent to the journal batcher.
pub enum JournalMessage {
    /// A journal entry to be batched and appended to WAL.
    Entry(JournalEntry),
    /// Flush all pending entries to WAL and signal completion.
    /// Used before orderbook snapshots to ensure snapshot is consistent with journal durability.
    Flush(oneshot::Sender<()>),
}

/// Sender half for pushing journal entries
pub type JournalBatchSender = mpsc::Sender<JournalMessage>;

enum ReplicationMessage {
    Entries(Vec<WalRecordWithOffset>),
    Shutdown,
    /// Barrier: replicate all pending entries, then fire the ack.
    /// Used by Flush to ensure the checkpoint is up-to-date before snapshot writes.
    FlushBarrier(oneshot::Sender<()>),
}

#[derive(Debug, Clone, Copy)]
struct ReplicationProgress {
    last_command_id: i64,
    max_l2_seq: i64,
}

#[cfg(feature = "rsm-state")]
struct RsmWalWriterState {
    commitments: ValidatorRsmStateCommitmentRuntime,
    next_batch_seq: u64,
    prev_block_hash: [u8; 32],
    global_command_seq: u64,
    block_signer: Option<alloy::signers::local::PrivateKeySigner>,
}

#[cfg(feature = "rsm-state")]
struct PreparedWalBlock {
    block: WalBlockRecordWithRoots,
    prepared_commitment: PreparedBatchCommitment,
}

#[cfg(feature = "rsm-state")]
impl RsmWalWriterState {
    fn prepare_block(&mut self, entries: &[JournalEntry]) -> Option<PreparedWalBlock> {
        let identity_hashes = engine_command_identity_hashes(entries);
        if identity_hashes.is_empty() {
            return None;
        }
        let timestamp = entries.last().map(|e| e.received_ts_ms).unwrap_or(0);
        let signer = self
            .block_signer
            .as_ref()
            .map(|signer| *signer.address().as_ref())
            .unwrap_or([0u8; 20]);
        let delta =
            rsm_state_delta_from_entries(entries, &identity_hashes, self.global_command_seq)
                .unwrap_or_else(|error| {
                    panic!(
                        "CRITICAL_FAILURE: failed to materialize validator RSM state delta: {}",
                        error
                    )
                });
        let prepared_commitment =
            self.commitments
                .prepare_rsm_batch(&delta)
                .unwrap_or_else(|error| {
                    panic!(
                        "CRITICAL_FAILURE: failed to prepare validator RSM state batch: {}",
                        error
                    )
                });
        let roots: RsmBatchRoots = prepared_commitment.commitment.clone().into();

        let mut block = WalBlockRecord::unsigned_from_identity_hashes(
            self.next_batch_seq,
            self.prev_block_hash,
            &identity_hashes,
            roots.batch_root,
            self.global_command_seq,
            timestamp,
            signer,
        );
        block.signature = self.sign_block(&block);

        Some(PreparedWalBlock {
            block: WalBlockRecordWithRoots {
                header: block,
                roots,
            },
            prepared_commitment,
        })
    }

    fn apply_prepared_block(&mut self, prepared: PreparedWalBlock) -> WalBlockRecordWithRoots {
        self.commitments
            .apply_prepared_rsm_batch(prepared.prepared_commitment)
            .unwrap_or_else(|error| {
                panic!(
                    "CRITICAL_FAILURE: failed to apply validator RSM state batch after WAL fsync: {}",
                    error
                )
            });
        self.advance_block_state(&prepared.block.header);
        prepared.block
    }

    fn sign_block(&self, unsigned: &WalBlockRecord) -> Vec<u8> {
        let Some(ref signer) = self.block_signer else {
            return Vec::new();
        };

        use alloy::primitives::B256;
        use alloy::signers::SignerSync;

        let hash = B256::from(unsigned.block_hash());
        match signer.sign_hash_sync(&hash) {
            Ok(sig) => {
                let mut sig_bytes = [0u8; 65];
                sig_bytes[..32].copy_from_slice(&sig.r().to_be_bytes::<32>());
                sig_bytes[32..64].copy_from_slice(&sig.s().to_be_bytes::<32>());
                sig_bytes[64] = sig.v() as u8;
                sig_bytes.to_vec()
            }
            Err(e) => {
                tracing::error!("Failed to sign block {}: {}", unsigned.batch_seq, e);
                Vec::new()
            }
        }
    }

    fn advance_block_state(&mut self, block: &WalBlockRecord) {
        let block_hash = block.block_hash();
        let signed = !block.signature.is_empty();
        tracing::debug!(
            batch_seq = block.batch_seq,
            command_count = block.command_count,
            first_seq = block.first_seq,
            block_hash = hex::encode(block_hash),
            signed,
            "Block produced"
        );

        self.prev_block_hash = block_hash;
        self.next_batch_seq += 1;
        self.global_command_seq += block.command_count;

        counter!("ht_blocks_produced_total").increment(1);
    }
}

/// The batcher that runs in the background and appends to local WAL first,
/// then replicates to PostgreSQL asynchronously.
pub struct EngineJournalBatcher {
    pool: crate::db_handler::DbPool,
    receiver: mpsc::Receiver<JournalMessage>,
    config: JournalBatcherConfig,
    shutdown: tokio::sync::broadcast::Receiver<()>,
    /// Sender half of the bounded channel to the dedicated WAL writer thread.
    wal_writer_tx: std::sync::mpsc::SyncSender<WalWriterMessage>,
    /// Join handle for the WAL writer OS thread (taken during shutdown).
    wal_writer_handle: Option<std::thread::JoinHandle<()>>,
    replication_tx: mpsc::Sender<ReplicationMessage>,
    replication_rx: Option<mpsc::Receiver<ReplicationMessage>>,
    /// Environment partition for persisted RSM roots and block headers.
    #[cfg(feature = "rsm-state")]
    rsm_environment: hypercall_db::ValidatorRsmEnvironment,
}

impl EngineJournalBatcher {
    /// Create a new batcher and return the sender for pushing entries.
    pub fn new(
        pool: crate::db_handler::DbPool,
        config: JournalBatcherConfig,
        shutdown: tokio::sync::broadcast::Receiver<()>,
        cold_signer: Option<alloy::signers::local::PrivateKeySigner>,
    ) -> (Self, JournalBatchSender) {
        let (tx, rx) = mpsc::channel(config.channel_capacity);

        // Create WalState directly — no Arc/Mutex needed, the writer thread owns it exclusively.
        let mut wal_state = WalState::open(config.wal_path.clone()).unwrap_or_else(|e| {
            panic!(
                "CRITICAL_FAILURE: Failed to initialize WAL state: {}. Restart required.",
                e
            )
        });

        #[cfg(feature = "rsm-state")]
        let rsm_environment = config.rsm_environment;
        #[cfg(feature = "rsm-state")]
        let mut commitments = ValidatorRsmStateCommitmentRuntime::open_from_db(
            config.rsm_commitment_store.clone(),
            Arc::new(pool.clone()),
            rsm_environment,
        )
        .unwrap_or_else(|error| {
            panic!(
                "CRITICAL_FAILURE: failed to initialize validator RSM state store: {}",
                error
            )
        });

        // Run WAL recovery synchronously before spawning the writer thread.
        // If recovery truncated torn bytes, fix up the write position so the
        // writer thread appends at the correct offset.
        if let Some(truncated_len) = Self::recover_wal_sync(
            &pool,
            &wal_state.wal_path,
            &wal_state.checkpoint_path,
            config.persist_digests,
            config.max_batch_size,
            #[cfg(feature = "rsm-state")]
            config.rsm_environment,
            #[cfg(feature = "rsm-state")]
            Some(&mut commitments),
        ) {
            wal_state.next_offset = truncated_len;
            wal_state.preallocated_end = truncated_len;
        }

        let (replication_tx, replication_rx) = mpsc::channel(config.channel_capacity);

        // Bounded sync channel for batcher → writer thread communication.
        let (wal_writer_tx, wal_writer_rx) = std::sync::mpsc::sync_channel(config.channel_capacity);

        // Set up hot key delegation if cold signer is provided.
        #[cfg(feature = "rsm-state")]
        let mut block_signer = None;
        #[cfg(feature = "rsm-state")]
        if let Some(cold_signer) = cold_signer {
            use alloy::primitives::B256;
            use alloy::signers::SignerSync;

            let hot_signer = alloy::signers::local::PrivateKeySigner::random();
            let hot_address = hot_signer.address();
            let cold_address = cold_signer.address();

            let now_ms = std::time::SystemTime::now()
                .duration_since(std::time::UNIX_EPOCH)
                .unwrap()
                .as_millis() as u64;
            let valid_until = now_ms + 24 * 60 * 60 * 1000; // 24 hours

            let mut authority = SigningAuthority {
                delegator: *cold_address.as_ref(),
                delegate: *hot_address.as_ref(),
                valid_from: now_ms,
                valid_until,
                signature: Vec::new(),
            };

            let delegation_hash = B256::from(authority.delegation_hash());
            match cold_signer.sign_hash_sync(&delegation_hash) {
                Ok(sig) => {
                    let mut sig_bytes = [0u8; 65];
                    sig_bytes[..32].copy_from_slice(&sig.r().to_be_bytes::<32>());
                    sig_bytes[32..64].copy_from_slice(&sig.s().to_be_bytes::<32>());
                    sig_bytes[64] = sig.v() as u8;
                    authority.signature = sig_bytes.to_vec();

                    info!(
                        cold_key = %cold_address,
                        hot_key = %hot_address,
                        valid_until_h = (valid_until - now_ms) / 3_600_000,
                        "Block signing: cold key delegated to ephemeral hot key (24h)"
                    );

                    block_signer = Some(hot_signer);
                }
                Err(e) => {
                    error!(
                        "Failed to sign delegation with cold key: {}. Blocks will be unsigned.",
                        e
                    );
                }
            }
        }
        #[cfg(not(feature = "rsm-state"))]
        let _ = cold_signer;
        #[cfg(feature = "rsm-state")]
        let (next_batch_seq, prev_block_hash, global_command_seq) =
            Self::load_rsm_block_cursor(&pool, rsm_environment);

        #[cfg(feature = "rsm-state")]
        let rsm_writer_state = RsmWalWriterState {
            commitments,
            next_batch_seq,
            prev_block_hash,
            global_command_seq,
            block_signer,
        };

        // Spawn dedicated WAL writer OS thread. It owns WalState and the RSM
        // commitment runtime so RocksDB mutation happens after WAL fsync.
        let replication_tx_for_thread = replication_tx.clone();
        let wal_writer_handle = std::thread::Builder::new()
            .name("wal-writer".to_string())
            .spawn(move || {
                wal_writer_thread_main(
                    wal_state,
                    wal_writer_rx,
                    replication_tx_for_thread,
                    #[cfg(feature = "rsm-state")]
                    rsm_writer_state,
                );
            })
            .expect("CRITICAL_FAILURE: failed to spawn wal-writer thread");

        let batcher = Self {
            pool,
            receiver: rx,
            config,
            shutdown,
            wal_writer_tx,
            wal_writer_handle: Some(wal_writer_handle),
            replication_tx,
            replication_rx: Some(replication_rx),
            #[cfg(feature = "rsm-state")]
            rsm_environment,
        };
        (batcher, tx)
    }

    #[cfg(feature = "rsm-state")]
    fn load_rsm_block_cursor(
        pool: &crate::db_handler::DbPool,
        rsm_environment: hypercall_db::ValidatorRsmEnvironment,
    ) -> (u64, [u8; 32], u64) {
        let handler =
            hypercall_db_diesel::DatabaseHandler::with_pool_no_migrations(Arc::new(pool.clone()));
        match handler.get_latest_rsm_block_sync(rsm_environment) {
            Ok(Some(block)) => {
                let next_batch_seq = block.block.height + 1;
                let prev_block_hash = block.block.hash;
                let global_command_seq = block.block.last_command_seq + 1;
                info!(
                    environment = %rsm_environment,
                    next_batch_seq = next_batch_seq,
                    global_command_seq = global_command_seq,
                    "Loaded validator RSM block cursor"
                );
                (next_batch_seq, prev_block_hash, global_command_seq)
            }
            Ok(None) => {
                info!(
                    environment = %rsm_environment,
                    "No validator RSM block cursor found; starting at genesis"
                );
                (0, [0u8; 32], 0)
            }
            Err(error) => {
                panic!(
                    "CRITICAL_FAILURE: failed to load validator RSM block cursor: {}",
                    error
                );
            }
        }
    }

    /// Run streaming WAL recovery. Returns `Some(truncated_len)` if the WAL
    /// was truncated to remove torn bytes, so the caller can fix up WalState.
    fn recover_wal_sync(
        pool: &crate::db_handler::DbPool,
        wal_path: &Path,
        checkpoint_path: &Path,
        persist_digests: bool,
        max_batch_size: usize,
        #[cfg(feature = "rsm-state")] rsm_environment: hypercall_db::ValidatorRsmEnvironment,
        #[cfg(feature = "rsm-state")] mut rsm_commitments: Option<
            &mut ValidatorRsmStateCommitmentRuntime,
        >,
    ) -> Option<u64> {
        // Stream WAL recovery in bounded-memory chunks instead of loading the
        // entire unreplicated tail into a Vec.  This prevents OOM when the
        // unreplicated tail is much larger than available RAM.

        if !wal_path.exists() {
            return None;
        }

        let checkpoint = read_checkpoint(checkpoint_path).unwrap_or_else(|e| {
            panic!(
                "CRITICAL_FAILURE: Failed to read WAL checkpoint at startup: {}. Restart required.",
                e
            )
        });

        // Open read-write so we can truncate torn bytes at EOF.
        let mut file = OpenOptions::new()
            .read(true)
            .write(true)
            .open(wal_path)
            .unwrap_or_else(|e| {
                panic!(
                    "CRITICAL_FAILURE: Failed to open WAL {} for replay: {}. Restart required.",
                    wal_path.display(),
                    e
                )
            });

        let file_len = file
            .metadata()
            .unwrap_or_else(|e| {
                panic!(
                    "CRITICAL_FAILURE: Failed to stat WAL {}: {}",
                    wal_path.display(),
                    e
                )
            })
            .len();

        if checkpoint.wal_offset > file_len {
            panic!(
                "CRITICAL_FAILURE: WAL checkpoint {} exceeds file length {} for {}",
                checkpoint.wal_offset,
                file_len,
                wal_path.display()
            );
        }

        if checkpoint.wal_offset == file_len {
            // Nothing to recover -- checkpoint is at end of WAL.
            return None;
        }

        file.seek(SeekFrom::Start(checkpoint.wal_offset))
            .unwrap_or_else(|e| panic!("CRITICAL_FAILURE: Failed to seek WAL replay start: {}", e));

        info!(
            wal_path = %wal_path.display(),
            checkpoint_path = %checkpoint_path.display(),
            checkpoint_offset = checkpoint.wal_offset,
            file_len = file_len,
            "Streaming WAL recovery to PostgreSQL before engine startup"
        );

        let batch_size = max_batch_size.max(1);
        let mut offset = checkpoint.wal_offset;
        // Tracks the offset after the last fully-parsed record so we can
        // truncate torn bytes that would prevent future records from being read.
        let mut last_good_offset = checkpoint.wal_offset;
        let mut total_recovered: u64 = 0;
        let mut hit_torn_record = false;

        // Outer loop: read and process one batch at a time.
        loop {
            let mut batch: Vec<WalRecordWithOffset> = Vec::with_capacity(batch_size);

            // Inner loop: fill one batch of up to `batch_size` commands.
            for _ in 0..batch_size {
                if batch.len() >= batch_size {
                    break;
                }

                let record_start_offset = offset;

                // Read 4-byte length prefix.
                let mut len_buf = [0u8; 4];
                match file.read_exact(&mut len_buf) {
                    Ok(()) => {}
                    Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                        hit_torn_record = true;
                        break;
                    }
                    Err(e) => {
                        panic!(
                            "CRITICAL_FAILURE: Failed reading WAL length prefix at offset {}: {}",
                            offset, e
                        );
                    }
                }

                let payload_len = u32::from_le_bytes(len_buf) as usize;
                if payload_len == 0 {
                    panic!(
                        "CRITICAL_FAILURE: WAL record length was zero at offset {} in {}. Data corruption detected.",
                        offset,
                        wal_path.display()
                    );
                }
                offset += 4;

                // Guard: a corrupt length prefix could claim gigabytes and OOM
                // the process before read_exact even gets a chance to fail.
                let remaining = file_len.saturating_sub(offset);
                if payload_len as u64 > remaining {
                    warn!(
                        "WAL declared payload_len={} but only {} bytes remain at offset {} in {}, \
                         truncating tail and replaying prior durable records only",
                        payload_len,
                        remaining,
                        offset,
                        wal_path.display(),
                    );
                    hit_torn_record = true;
                    break;
                }

                // Read 4-byte CRC.
                let mut crc_buf = [0u8; 4];
                match file.read_exact(&mut crc_buf) {
                    Ok(()) => {}
                    Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                        warn!(
                            "WAL ended with incomplete CRC header at offset {} in {} ({}), \
                             truncating tail and replaying prior durable records only",
                            offset,
                            wal_path.display(),
                            e
                        );
                        hit_torn_record = true;
                        break;
                    }
                    Err(e) => {
                        panic!(
                            "CRITICAL_FAILURE: Failed reading WAL CRC prefix at offset {}: {}",
                            offset, e
                        );
                    }
                }
                offset += 4;

                let expected_crc = u32::from_le_bytes(crc_buf);
                let mut payload = vec![0u8; payload_len];

                // Read payload.
                match file.read_exact(&mut payload) {
                    Ok(()) => {}
                    Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                        warn!(
                            "WAL ended with incomplete payload at offset {} in {} ({}), \
                             truncating tail and replaying prior durable records only",
                            offset,
                            wal_path.display(),
                            e
                        );
                        hit_torn_record = true;
                        break;
                    }
                    Err(e) => {
                        panic!(
                            "CRITICAL_FAILURE: Failed reading WAL payload at offset {}: {}",
                            offset, e
                        );
                    }
                }
                offset += payload_len as u64;

                let actual_crc = WAL_CRC.checksum(&payload);
                if actual_crc != expected_crc {
                    panic!(
                        "CRITICAL_FAILURE: WAL CRC mismatch at end_offset {} in {}: expected {}, got {}. Data corruption detected.",
                        offset,
                        wal_path.display(),
                        expected_crc,
                        actual_crc
                    );
                }

                let mut decoded_records = Vec::new();
                #[cfg(feature = "rsm-state")]
                if let Ok(block_entry) = rmp_serde::from_slice::<WalDurableBlockEntry>(&payload) {
                    let header = block_entry.header;
                    let roots = block_entry.roots;
                    for (index, cmd) in block_entry.commands.into_iter().enumerate() {
                        decoded_records.push(WalRecordWithOffset {
                            record: cmd,
                            end_offset: offset,
                            block_header: (index == 0).then_some(header.clone()),
                            block_roots: (index == 0).then_some(roots),
                        });
                    }
                } else if rmp_serde::from_slice::<WalBlockEntry>(&payload).is_ok() {
                    panic!(
                        "CRITICAL_FAILURE: legacy rootless RSM WAL block at end_offset {} is unsupported after durable root cutover",
                        offset
                    );
                } else {
                    match rmp_serde::from_slice::<WalJournalRecord>(&payload) {
                        Ok(record) => {
                            decoded_records.push(WalRecordWithOffset {
                                record,
                                end_offset: offset,
                                block_header: None,
                                block_roots: None,
                            });
                        }
                        Err(_) => {
                            if rmp_serde::from_slice::<WalBlockRecord>(&payload).is_ok() {
                                tracing::debug!(
                                    end_offset = offset,
                                    "Skipping legacy WAL block record during recovery"
                                );
                                last_good_offset = offset;
                                continue;
                            }

                            panic!(
                                "CRITICAL_FAILURE: Failed to deserialize WAL payload at end_offset {}",
                                offset
                            );
                        }
                    }
                }
                #[cfg(not(feature = "rsm-state"))]
                match rmp_serde::from_slice::<WalJournalRecord>(&payload) {
                    Ok(record) => {
                        decoded_records.push(WalRecordWithOffset {
                            record,
                            end_offset: offset,
                        });
                    }
                    Err(_) => {
                        panic!(
                            "CRITICAL_FAILURE: Failed to deserialize WAL payload at end_offset {}",
                            offset
                        );
                    }
                }

                if should_defer_replay_record(batch.len(), decoded_records.len(), batch_size) {
                    file.seek(SeekFrom::Start(record_start_offset))
                        .unwrap_or_else(|e| {
                            panic!(
                                "CRITICAL_FAILURE: Failed to rewind WAL replay batch boundary: {}",
                                e
                            )
                        });
                    offset = record_start_offset;
                    break;
                }

                if decoded_records.len() > batch_size {
                    warn!(
                        commands = decoded_records.len(),
                        batch_size = batch_size,
                        end_offset = offset,
                        "WAL replay physical record exceeds command batch size; replaying it atomically"
                    );
                }

                for record in decoded_records {
                    batch.push(WalRecordWithOffset {
                        record: record.record,
                        end_offset: record.end_offset,
                        #[cfg(feature = "rsm-state")]
                        block_header: record.block_header,
                        #[cfg(feature = "rsm-state")]
                        block_roots: record.block_roots,
                    });
                }

                last_good_offset = offset;
            }

            // If no records were read in this iteration, we are done.
            if batch.is_empty() {
                break;
            }

            let batch_len = batch.len() as u64;

            #[cfg(feature = "rsm-state")]
            if let Some(commitments) = rsm_commitments.as_deref_mut() {
                apply_recovered_rsm_commitment_blocks(&mut batch, commitments).unwrap_or_else(
                    |error| {
                        panic!(
                            "CRITICAL_FAILURE: WAL replay failed to rebuild local validator RSM state before root persistence: {}",
                            error
                        )
                    },
                );
            }

            // Insert this chunk into PostgreSQL.
            let db_entries: Vec<JournalEntry> =
                batch.iter().map(WalRecordWithOffset::to_db_entry).collect();
            let (_inserted_count, mut inserted_commands) = match Self::insert_batch(
                pool,
                &db_entries,
                persist_digests,
                #[cfg(feature = "rsm-state")]
                Some(
                    Self::rsm_block_persistence_batch(RsmBlockPersistenceInput {
                        environment: rsm_environment,
                        records: &batch,
                        mode: RsmBlockPersistenceMode::Replay,
                    })
                    .unwrap_or_else(|error| {
                        panic!(
                            "CRITICAL_FAILURE: WAL replay failed to prepare RSM block persistence: {}",
                            error
                        )
                    }),
                ),
            ) {
                Ok(result) => result,
                Err(e) => {
                    panic!(
                        "CRITICAL_FAILURE: WAL replay insert failed: {}. Startup cannot continue.",
                        e
                    );
                }
            };

            if inserted_commands.is_empty() {
                let request_ids: Vec<hypercall_db_diesel::engine_enums::DbUuid> = batch
                    .iter()
                    .map(|record| record.record.request_uuid)
                    .collect();
                inserted_commands =
                    Self::lookup_command_ids_by_request_ids(pool, &request_ids).unwrap_or_else(
                        |e| {
                            panic!(
                                "CRITICAL_FAILURE: WAL replay command_id lookup failed: {}. Startup cannot continue.",
                                e
                            )
                        },
                    );
                if inserted_commands.is_empty() {
                    panic!(
                        "CRITICAL_FAILURE: WAL replay insert and command_id lookup returned empty for non-empty chunk."
                    );
                }
            }

            let progress = ReplicationProgress {
                last_command_id: inserted_commands
                    .iter()
                    .map(|(_, command_id)| *command_id)
                    .max()
                    .expect("WAL replay command IDs unexpectedly empty"),
                max_l2_seq: batch
                    .iter()
                    .flat_map(|record| record.record.events.iter().filter_map(|e| e.l2_sequence))
                    .max()
                    .unwrap_or(0),
            };

            let last_offset = batch
                .last()
                .expect("WAL replay chunk unexpectedly empty")
                .end_offset;

            // Advance checkpoint after each chunk so crash-recovery resumes here.
            if let Err(e) = Self::advance_checkpoint(checkpoint_path, last_offset, progress) {
                panic!(
                    "CRITICAL_FAILURE: Failed to advance WAL checkpoint after replay: {}",
                    e
                );
            }

            total_recovered += batch_len;

            // Drop the batch to free memory before reading the next chunk.
            drop(batch);

            debug!(
                total_recovered = total_recovered,
                current_offset = offset,
                file_len = file_len,
                "WAL streaming recovery progress"
            );
        }

        // Truncate torn bytes so the writer thread appends cleanly after the
        // last complete record.  Without this, the partial bytes sit between
        // the checkpoint and the next write, and the next restart would re-parse
        // (and skip) them before any new records could be reached.
        let truncated = if hit_torn_record && last_good_offset < file_len {
            warn!(
                "Truncating WAL {} from {} to {} bytes to remove torn tail",
                wal_path.display(),
                file_len,
                last_good_offset
            );
            file.set_len(last_good_offset).unwrap_or_else(|e| {
                panic!(
                    "CRITICAL_FAILURE: Failed to truncate WAL {} to {}: {}",
                    wal_path.display(),
                    last_good_offset,
                    e
                )
            });
            Some(last_good_offset)
        } else {
            None
        };

        if total_recovered > 0 {
            info!(
                total_recovered = total_recovered,
                "WAL streaming recovery complete"
            );
        }

        counter!("ht_journal_wal_replay_entries_total").increment(total_recovered);
        truncated
    }

    /// Run the batcher loop with group commit.
    ///
    /// Entries are flushed immediately on arrival. Natural batching occurs when
    /// new entries arrive while the writer thread is performing fsync — they
    /// accumulate in the channel and get drained in the next iteration.
    pub async fn run(mut self) {
        info!(
            "Starting WAL-first journal batcher (group_commit=true, batch_size={}, digests={})",
            self.config.max_batch_size, self.config.persist_digests
        );

        let flush_interval = Duration::from_millis(self.config.flush_interval_ms);
        let wal_path = self.config.wal_path.clone();
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        let replication_rx = self
            .replication_rx
            .take()
            .expect("Replication receiver missing; batcher initialized incorrectly");

        let replication_handle = tokio::spawn(Self::replication_loop(
            self.pool.clone(),
            self.config.persist_digests,
            self.config.max_batch_size,
            flush_interval,
            checkpoint_path,
            wal_path,
            replication_rx,
            #[cfg(feature = "rsm-state")]
            self.rsm_environment,
        ));

        let mut batch: Vec<JournalEntry> = Vec::with_capacity(self.config.max_batch_size);

        loop {
            tokio::select! {
                biased;

                _ = self.shutdown.recv() => {
                    info!("Journal batcher received shutdown, flushing remaining {} entries", batch.len());
                    if !batch.is_empty() {
                        self.flush_batch(&mut batch, None).await;
                    }

                    // CRITICAL: Drain all remaining entries from receiver to WAL.
                    let mut drained_count = 0;
                    while let Ok(msg) = self.receiver.try_recv() {
                        match msg {
                            JournalMessage::Entry(entry) => {
                                drained_count += 1;
                                batch.push(entry);
                                if batch.len() >= self.config.max_batch_size {
                                    self.flush_batch(&mut batch, None).await;
                                }
                            }
                            JournalMessage::Flush(ack) => {
                                let (complete_tx, complete_rx) = oneshot::channel();
                                if !batch.is_empty() {
                                    self.flush_batch(&mut batch, Some(complete_tx)).await;
                                } else {
                                    self.send_flush_barrier(complete_tx).await;
                                }
                                let _ = complete_rx.await;
                                self.await_replication_flush().await;
                                let _ = ack.send(());
                            }
                        }
                    }

                    if !batch.is_empty() {
                        // Final flush with completion to ensure durability before shutdown.
                        let (complete_tx, complete_rx) = oneshot::channel();
                        self.flush_batch(&mut batch, Some(complete_tx)).await;
                        let _ = complete_rx.await;
                    }

                    if drained_count > 0 {
                        info!("Drained {} additional entries to WAL during shutdown", drained_count);
                    }

                    // Shutdown WAL writer thread.
                    let _ = self.wal_writer_tx.send(WalWriterMessage::Shutdown);
                    if let Some(handle) = self.wal_writer_handle.take() {
                        handle.join().expect("CRITICAL_FAILURE: wal-writer thread panicked");
                    }

                    if let Err(e) = self.replication_tx.send(ReplicationMessage::Shutdown).await {
                        warn!("Failed to signal replication shutdown: {}", e);
                    }

                    match replication_handle.await {
                        Ok(()) => {}
                        Err(e) => {
                            panic!(
                                "CRITICAL_FAILURE: Journal replication task panicked during shutdown: {}",
                                e
                            );
                        }
                    }

                    break;
                }

                msg = self.receiver.recv() => {
                    match msg {
                        Some(JournalMessage::Entry(e)) => {
                            // Track queue latency (time from engine push to batcher receive)
                            let queue_latency = e.created_at.elapsed();
                            histogram!("ht_journal_queue_latency_seconds")
                                .record(queue_latency.as_secs_f64());

                            batch.push(e);

                            // Group commit: drain all immediately available entries.
                            // Entries that arrived while the previous fsync was in flight
                            // get batched together, amortizing the cost of the next fsync.
                            while batch.len() < self.config.max_batch_size {
                                match self.receiver.try_recv() {
                                    Ok(JournalMessage::Entry(e2)) => {
                                        let queue_latency = e2.created_at.elapsed();
                                        histogram!("ht_journal_queue_latency_seconds")
                                            .record(queue_latency.as_secs_f64());
                                        batch.push(e2);
                                    }
                                    Ok(JournalMessage::Flush(ack)) => {
                                        // Flush arrived during drain: flush current batch
                                        // and wait for durability before acking.
                                        let (complete_tx, complete_rx) = oneshot::channel();
                                        if !batch.is_empty() {
                                            self.flush_batch(&mut batch, Some(complete_tx)).await;
                                        } else {
                                            self.send_flush_barrier(complete_tx).await;
                                        }
                                        let _ = complete_rx.await;
                                        self.await_replication_flush().await;
                                        let _ = ack.send(());
                                    }
                                    Err(_) => break,
                                }
                            }

                            // Immediate flush — no timer delay.
                            if !batch.is_empty() {
                                self.flush_batch(&mut batch, None).await;
                            }
                        }
                        Some(JournalMessage::Flush(ack)) => {
                            // Always send through the writer thread to serialize
                            // behind any in-flight WriteBatch fsyncs.
                            let (complete_tx, complete_rx) = oneshot::channel();
                            if !batch.is_empty() {
                                self.flush_batch(&mut batch, Some(complete_tx)).await;
                            } else {
                                self.send_flush_barrier(complete_tx).await;
                            }
                            let _ = complete_rx.await;
                            // Wait for replication to complete so the checkpoint
                            // is current before any snapshot write.
                            self.await_replication_flush().await;
                            let _ = ack.send(());
                        }
                        None => {
                            info!("Journal batcher channel closed, flushing remaining {} entries", batch.len());
                            if !batch.is_empty() {
                                self.flush_batch(&mut batch, None).await;
                            }
                            // Shutdown WAL writer thread.
                            let _ = self.wal_writer_tx.send(WalWriterMessage::Shutdown);
                            if let Some(handle) = self.wal_writer_handle.take() {
                                handle.join().expect("CRITICAL_FAILURE: wal-writer thread panicked");
                            }
                            let _ = self.replication_tx.send(ReplicationMessage::Shutdown).await;
                            match replication_handle.await {
                                Ok(()) => {}
                                Err(e) => {
                                    panic!(
                                        "CRITICAL_FAILURE: Journal replication task panicked while closing channel: {}",
                                        e
                                    );
                                }
                            }
                            break;
                        }
                    }
                }
            }
        }

        info!("Journal batcher stopped");
    }

    /// Enqueue a batch for the WAL writer thread.
    ///
    /// Extracts ack senders from the entries and sends a `WalWriterMessage::WriteBatch`
    /// to the writer thread. The writer thread handles write + fsync + ack firing.
    ///
    /// If `completion` is Some, the writer thread fires it after fsync. Callers
    /// who need to block until fsync (e.g. Flush) should await the receiver side.
    async fn flush_batch(
        &mut self,
        batch: &mut Vec<JournalEntry>,
        completion: Option<oneshot::Sender<()>>,
    ) {
        if batch.is_empty() {
            return;
        }

        let batch_size = batch.len();
        let mut entries = std::mem::take(batch);

        histogram!("ht_journal_batch_size").record(batch_size as f64);

        let ack_senders: Vec<Option<oneshot::Sender<()>>> =
            entries.iter_mut().map(|e| e.commit_ack.take()).collect();

        let msg = WalWriterMessage::WriteBatch {
            entries,
            ack_senders,
            completion,
        };

        // Send to writer thread via bounded sync channel.
        // Use spawn_blocking because SyncSender::send blocks if full (backpressure).
        let tx = self.wal_writer_tx.clone();
        tokio::task::spawn_blocking(move || {
            tx.send(msg)
                .expect("CRITICAL_FAILURE: wal-writer thread dead. Restart required.");
        })
        .await
        .expect("CRITICAL_FAILURE: spawn_blocking failed sending to wal-writer");
    }

    /// Send a flush barrier to the WAL writer thread.
    ///
    /// The barrier is processed in FIFO order behind any in-flight WriteBatch
    /// messages, ensuring the completion fires only after all prior fsyncs.
    async fn send_flush_barrier(&self, completion: oneshot::Sender<()>) {
        let tx = self.wal_writer_tx.clone();
        tokio::task::spawn_blocking(move || {
            tx.send(WalWriterMessage::FlushBarrier { completion })
                .expect("CRITICAL_FAILURE: wal-writer thread dead. Restart required.");
        })
        .await
        .expect("CRITICAL_FAILURE: spawn_blocking failed sending flush barrier");
    }

    /// Send a flush barrier through the replication channel and wait for all
    /// pending WAL entries to be replicated to Postgres.
    ///
    /// After this returns, the WAL checkpoint is guaranteed to reflect the state
    /// of all entries written to the WAL up to this point. This prevents
    /// snapshot-checkpoint lag: without this barrier, `persist_engine_state_snapshot`
    /// could read a stale checkpoint and tag the snapshot with a `last_command_id`
    /// that trails the live engine state, causing double-application of fills
    /// during the next recovery.
    async fn await_replication_flush(&self) {
        let (tx, rx) = oneshot::channel();
        if let Err(e) = self
            .replication_tx
            .send(ReplicationMessage::FlushBarrier(tx))
            .await
        {
            warn!(
                "Failed to send replication flush barrier (channel closed during shutdown): {}",
                e
            );
            return;
        }
        match rx.await {
            Ok(()) => {
                debug!("Replication flush barrier completed — checkpoint is up-to-date");
            }
            Err(_) => {
                warn!("Replication flush barrier ack dropped (replication task exited)");
            }
        }
    }

    /// Replication loop: drains WAL records from the writer thread and replicates
    /// to PostgreSQL.
    ///
    /// NOTE: This loop does NOT listen to the broadcast shutdown signal directly.
    /// Shutdown is driven by `ReplicationMessage::Shutdown` sent by the batcher
    /// AFTER the WAL writer thread has been joined. This prevents a race where
    /// the broadcast fires early, the replication loop exits, and the WAL writer
    /// thread panics on `replication_tx.blocking_send()` to a closed channel.
    async fn replication_loop(
        pool: crate::db_handler::DbPool,
        persist_digests: bool,
        max_batch_size: usize,
        flush_interval: Duration,
        checkpoint_path: PathBuf,
        wal_path: PathBuf,
        mut receiver: mpsc::Receiver<ReplicationMessage>,
        #[cfg(feature = "rsm-state")] rsm_environment: hypercall_db::ValidatorRsmEnvironment,
    ) {
        info!("Journal replication loop started");

        let mut pending: Vec<WalRecordWithOffset> = Vec::new();
        let mut last_flush = Instant::now();
        let mut last_punched_offset: u64 =
            Self::read_checkpoint_or_panic(&checkpoint_path).wal_offset;
        Self::emit_wal_checkpoint_metrics(&wal_path, last_punched_offset);

        loop {
            tokio::select! {
                biased;

                msg = receiver.recv() => {
                    match msg {
                        Some(ReplicationMessage::Entries(entries)) => {
                            pending.extend(entries);
                            gauge!("ht_journal_replication_backlog").set(pending.len() as f64);
                            if pending.len() >= max_batch_size {
                                if let Err(e) = Self::replicate_pending(
                                    &pool,
                                    persist_digests,
                                    max_batch_size,
                                    &checkpoint_path,
                                    &mut pending,
                                    #[cfg(feature = "rsm-state")]
                                    rsm_environment,
                                ).await {
                                    // pending is intentionally not drained; entries will be retried on next flush
                                    error!(
                                        pending = pending.len(),
                                        "Replication batch failed (WAL tail will grow): {}",
                                        e
                                    );
                                    counter!("ht_journal_replication_batch_failures_total").increment(1);
                                } else {
                                    last_flush = Instant::now();
                                    let cp = Self::read_checkpoint_or_panic(&checkpoint_path);
                                    Self::emit_wal_checkpoint_metrics(&wal_path, cp.wal_offset);
                                    if cp.wal_offset > last_punched_offset {
                                        Self::punch_wal_hole(&wal_path, last_punched_offset, cp.wal_offset);
                                        last_punched_offset = cp.wal_offset;
                                    }
                                }
                            }
                        }
                        Some(ReplicationMessage::FlushBarrier(ack)) => {
                            // Replicate ALL pending entries before acking. This
                            // guarantees the checkpoint reflects everything the
                            // WAL writer has sent us so far. Unlike the normal
                            // replication path, we MUST succeed here — a stale
                            // checkpoint would cause snapshot-checkpoint lag.
                            if !pending.is_empty() {
                                if let Err(e) = Self::replicate_pending(
                                    &pool,
                                    persist_digests,
                                    max_batch_size,
                                    &checkpoint_path,
                                    &mut pending,
                                    #[cfg(feature = "rsm-state")]
                                    rsm_environment,
                                ).await {
                                    error!(
                                        "CRITICAL_FAILURE: replication flush barrier failed: {}. \
                                         Cannot write snapshot with stale checkpoint. Aborting process.",
                                        e
                                    );
                                    std::process::abort();
                                }
                                last_flush = Instant::now();
                                let cp = Self::read_checkpoint_or_panic(&checkpoint_path);
                                Self::emit_wal_checkpoint_metrics(&wal_path, cp.wal_offset);
                                if cp.wal_offset > last_punched_offset {
                                    Self::punch_wal_hole(&wal_path, last_punched_offset, cp.wal_offset);
                                    last_punched_offset = cp.wal_offset;
                                }
                            }
                            let _ = ack.send(());
                        }
                        Some(ReplicationMessage::Shutdown) => {
                            info!("Journal replication received shutdown with {} pending entries", pending.len());
                            Self::drain_replication_channel(&mut receiver, &mut pending);
                            if let Err(e) = Self::replicate_pending(
                                &pool,
                                persist_digests,
                                max_batch_size,
                                &checkpoint_path,
                                &mut pending,
                                #[cfg(feature = "rsm-state")]
                                rsm_environment,
                            ).await {
                                error!("Replication failed during shutdown (data safe in WAL): {}", e);
                                counter!("ht_journal_replication_batch_failures_total").increment(1);
                            }
                            break;
                        }
                        None => {
                            info!("Replication channel closed with {} pending entries", pending.len());
                            if let Err(e) = Self::replicate_pending(
                                &pool,
                                persist_digests,
                                max_batch_size,
                                &checkpoint_path,
                                &mut pending,
                                #[cfg(feature = "rsm-state")]
                                rsm_environment,
                            ).await {
                                error!("Replication failed on channel close (data safe in WAL): {}", e);
                                counter!("ht_journal_replication_batch_failures_total").increment(1);
                            }
                            break;
                        }
                    }
                }

                _ = tokio::time::sleep(flush_interval) => {
                    if !pending.is_empty() && last_flush.elapsed() >= flush_interval {
                        if let Err(e) = Self::replicate_pending(
                            &pool,
                            persist_digests,
                            max_batch_size,
                            &checkpoint_path,
                            &mut pending,
                            #[cfg(feature = "rsm-state")]
                            rsm_environment,
                        ).await {
                            // pending is intentionally not drained; entries will be retried on next flush
                            error!(
                                pending = pending.len(),
                                "Replication flush failed (WAL tail will grow): {}",
                                e
                            );
                            counter!("ht_journal_replication_batch_failures_total").increment(1);
                        } else {
                            last_flush = Instant::now();
                        }
                        let cp = Self::read_checkpoint_or_panic(&checkpoint_path);
                        Self::emit_wal_checkpoint_metrics(&wal_path, cp.wal_offset);
                        if cp.wal_offset > last_punched_offset {
                            Self::punch_wal_hole(&wal_path, last_punched_offset, cp.wal_offset);
                            last_punched_offset = cp.wal_offset;
                        }
                    }
                }
            }
        }

        info!("Journal replication loop stopped");
    }

    fn read_checkpoint_or_panic(path: &Path) -> WalCheckpointMetadata {
        read_checkpoint(path).unwrap_or_else(|e| {
            panic!(
                "CRITICAL_FAILURE: failed to read/parse checkpoint {}: {}",
                path.display(),
                e
            )
        })
    }

    fn drain_replication_channel(
        receiver: &mut mpsc::Receiver<ReplicationMessage>,
        pending: &mut Vec<WalRecordWithOffset>,
    ) {
        while let Ok(msg) = receiver.try_recv() {
            match msg {
                ReplicationMessage::Entries(entries) => pending.extend(entries),
                ReplicationMessage::FlushBarrier(ack) => {
                    // During drain, ack immediately — all pending entries will
                    // be replicated by the caller after drain completes.
                    let _ = ack.send(());
                }
                ReplicationMessage::Shutdown => break,
            }
        }
    }

    /// Punch a hole in the WAL file for the replicated byte range [from, to).
    /// Releases disk blocks without changing the file size (sparse file).
    /// Non-fatal: logs a warning if the filesystem doesn't support it.
    fn punch_wal_hole(wal_path: &Path, from: u64, to: u64) {
        if to <= from {
            return;
        }

        #[cfg(target_os = "linux")]
        {
            use std::os::unix::io::AsRawFd;
            let file = match OpenOptions::new().write(true).open(wal_path) {
                Ok(f) => f,
                Err(e) => {
                    warn!("Failed to open WAL for hole punch: {}", e);
                    return;
                }
            };
            let ret = unsafe {
                libc::fallocate(
                    file.as_raw_fd(),
                    libc::FALLOC_FL_PUNCH_HOLE | libc::FALLOC_FL_KEEP_SIZE,
                    from as i64,
                    (to - from) as i64,
                )
            };
            if ret == 0 {
                let reclaimed_mb = (to - from) / (1024 * 1024);
                counter!("ht_wal_hole_punch_bytes").increment(to - from);
                debug!(
                    "WAL hole punch: reclaimed ~{}MB (offset {}..{})",
                    reclaimed_mb, from, to
                );
            } else {
                let errno = std::io::Error::last_os_error();
                warn!(
                    "WAL hole punch failed (non-fatal): {} (offset {}..{})",
                    errno, from, to
                );
            }
        }

        #[cfg(not(target_os = "linux"))]
        {
            let _ = (wal_path, from, to);
        }
    }

    fn advance_checkpoint(
        checkpoint_path: &Path,
        last_offset: u64,
        progress: ReplicationProgress,
    ) -> Result<(), String> {
        let current = read_checkpoint(checkpoint_path)?;
        let next = WalCheckpointMetadata {
            wal_offset: last_offset,
            last_command_id: current.last_command_id.max(progress.last_command_id),
            last_l2_seq: current.last_l2_seq.max(progress.max_l2_seq),
        };
        write_checkpoint(checkpoint_path, next)
    }

    /// Emit Prometheus gauges for WAL checkpoint offset and unreplicated tail.
    fn emit_wal_checkpoint_metrics(wal_path: &Path, checkpoint_offset: u64) {
        gauge!("ht_wal_checkpoint_offset_bytes").set(checkpoint_offset as f64);
        match std::fs::metadata(wal_path) {
            Ok(meta) => {
                let unreplicated = meta.len().saturating_sub(checkpoint_offset);
                gauge!("ht_wal_unreplicated_bytes").set(unreplicated as f64);
            }
            Err(e) => {
                warn!("Failed to stat WAL for checkpoint metrics: {}", e);
            }
        }
    }

    async fn replicate_pending(
        pool: &crate::db_handler::DbPool,
        persist_digests: bool,
        max_batch_size: usize,
        checkpoint_path: &Path,
        pending: &mut Vec<WalRecordWithOffset>,
        #[cfg(feature = "rsm-state")] rsm_environment: hypercall_db::ValidatorRsmEnvironment,
    ) -> Result<(), String> {
        while !pending.is_empty() {
            let chunk_len = next_replication_chunk_len(pending, max_batch_size);
            let chunk: Vec<WalRecordWithOffset> = pending.iter().take(chunk_len).cloned().collect();

            match Self::replicate_chunk(
                pool,
                persist_digests,
                &chunk,
                #[cfg(feature = "rsm-state")]
                rsm_environment,
            )
            .await
            {
                Ok(progress) => {
                    let last_offset = chunk
                        .last()
                        .expect("replication chunk unexpectedly empty")
                        .end_offset;
                    Self::advance_checkpoint(checkpoint_path, last_offset, progress)?;
                    pending.drain(..chunk_len);
                    gauge!("ht_journal_replication_backlog").set(pending.len() as f64);
                    counter!("ht_journal_replication_chunks_total").increment(1);
                }
                Err(e) => {
                    counter!("ht_journal_replication_errors_total").increment(1);
                    warn!(
                        "Journal replication chunk failed, batch remains pending for next flush cycle (pending={}): {}",
                        pending.len(),
                        e
                    );
                    tokio::time::sleep(REPLICATION_RETRY_DELAY).await;
                    return Err(e);
                }
            }
        }

        Ok(())
    }

    async fn replicate_chunk(
        pool: &crate::db_handler::DbPool,
        persist_digests: bool,
        records: &[WalRecordWithOffset],
        #[cfg(feature = "rsm-state")] rsm_environment: hypercall_db::ValidatorRsmEnvironment,
    ) -> Result<ReplicationProgress, String> {
        if records.is_empty() {
            return Ok(ReplicationProgress {
                last_command_id: 0,
                max_l2_seq: 0,
            });
        }

        let db_entries: Vec<JournalEntry> = records
            .iter()
            .map(WalRecordWithOffset::to_db_entry)
            .collect();

        let pool_for_insert = pool.clone();
        #[cfg(feature = "rsm-state")]
        let block_records = records.to_vec();
        let insert_start = Instant::now();
        let insert_result = tokio::task::spawn_blocking(move || {
            Self::insert_batch(
                &pool_for_insert,
                &db_entries,
                persist_digests,
                #[cfg(feature = "rsm-state")]
                Some(Self::rsm_block_persistence_batch(
                    RsmBlockPersistenceInput {
                        environment: rsm_environment,
                        records: &block_records,
                        mode: RsmBlockPersistenceMode::Live,
                    },
                )?),
            )
        })
        .await
        .map_err(|e| format!("spawn_blocking failed during replication insert: {}", e))?;

        histogram!("ht_journal_postgres_insert_seconds")
            .record(insert_start.elapsed().as_secs_f64());

        let (_inserted_count, mut inserted_commands) =
            insert_result.map_err(|e| format!("replication insert failed: {}", e))?;

        if inserted_commands.is_empty() {
            // Insert may have succeeded on a previous attempt, so rehydrate command IDs.
            let request_ids: Vec<hypercall_db_diesel::engine_enums::DbUuid> =
                records.iter().map(|r| r.record.request_uuid).collect();
            let pool_for_lookup = pool.clone();
            let lookup_result = tokio::task::spawn_blocking(move || {
                Self::lookup_command_ids_by_request_ids(&pool_for_lookup, &request_ids)
            })
            .await
            .map_err(|e| format!("spawn_blocking failed during command_id lookup: {}", e))?;

            inserted_commands =
                lookup_result.map_err(|e| format!("command_id lookup failed: {}", e))?;

            if inserted_commands.is_empty() {
                return Err(
                    "replication insert returned no command IDs and lookup by request_id returned empty"
                        .to_string(),
                );
            }
        }

        let last_command_id = inserted_commands
            .iter()
            .map(|(_, command_id)| *command_id)
            .max()
            .expect("inserted_commands unexpectedly empty after rehydration");

        let max_l2_seq = records
            .iter()
            .flat_map(|record| record.record.events.iter().filter_map(|e| e.l2_sequence))
            .max()
            .unwrap_or(0);

        Ok(ReplicationProgress {
            last_command_id,
            max_l2_seq,
        })
    }

    /// Insert a batch of entries into PostgreSQL (blocking).
    /// Returns (inserted_count, [(request_uuid, command_id)]) for downstream event publish accounting.
    fn insert_batch(
        pool: &crate::db_handler::DbPool,
        entries: &[JournalEntry],
        persist_digests: bool,
        #[cfg(feature = "rsm-state")] rsm_blocks: Option<hypercall_db::EngineJournalRsmBlockBatch>,
    ) -> anyhow::Result<(usize, Vec<(hypercall_db_diesel::engine_enums::DbUuid, i64)>)> {
        let db_entries: Vec<_> = entries.iter().map(engine_journal_entry_insert).collect();
        let handler =
            hypercall_db_diesel::DatabaseHandler::with_pool_no_migrations(Arc::new(pool.clone()));
        #[cfg(feature = "rsm-state")]
        let rsm_blocks = rsm_blocks.as_ref();
        #[cfg(not(feature = "rsm-state"))]
        let rsm_blocks = None;
        let result =
            handler.insert_engine_journal_batch_sync(&db_entries, persist_digests, rsm_blocks)?;
        Ok((
            result.inserted_count,
            result
                .inserted_commands
                .into_iter()
                .map(|(request_uuid, command_id)| {
                    (
                        hypercall_db_diesel::engine_enums::DbUuid(request_uuid),
                        command_id,
                    )
                })
                .collect(),
        ))
    }

    fn lookup_command_ids_by_request_ids(
        pool: &crate::db_handler::DbPool,
        request_ids: &[hypercall_db_diesel::engine_enums::DbUuid],
    ) -> anyhow::Result<Vec<(hypercall_db_diesel::engine_enums::DbUuid, i64)>> {
        let request_ids: Vec<_> = request_ids.iter().map(|id| id.0).collect();
        let handler =
            hypercall_db_diesel::DatabaseHandler::with_pool_no_migrations(Arc::new(pool.clone()));
        Ok(handler
            .lookup_engine_journal_command_ids_sync(&request_ids)?
            .into_iter()
            .map(|(request_uuid, command_id)| {
                (
                    hypercall_db_diesel::engine_enums::DbUuid(request_uuid),
                    command_id,
                )
            })
            .collect())
    }

    /// Start the batcher as a background task
    pub fn start(self) -> tokio::task::JoinHandle<()> {
        tokio::spawn(
            async move {
                self.run().await;
            }
            .instrument(info_span!("journal_batcher")),
        )
    }
}

fn engine_journal_entry_insert(entry: &JournalEntry) -> hypercall_db::EngineJournalEntryInsert {
    let pre_digest_json = serde_json::to_value(&entry.pre_digest)
        .expect("Failed to serialize pre_digest for journal insert");
    let post_digest_json = serde_json::to_value(&entry.post_digest)
        .expect("Failed to serialize post_digest for journal insert");

    hypercall_db::EngineJournalEntryInsert {
        request_uuid: entry.request_uuid.0,
        received_ts_ms: entry.received_ts_ms,
        command_data: entry.command_data.clone(),
        response_data: entry.response_data.clone(),
        order_id: entry.order_id,
        command_type: entry
            .command_type_enum
            .map(|command| command.as_str().to_string()),
        duration_ms: entry.duration_ms,
        pre_digest_data: hypercall_types::serialize_to_wire_bytes(&pre_digest_json),
        post_digest_data: hypercall_types::serialize_to_wire_bytes(&post_digest_json),
        events: entry
            .events
            .iter()
            .map(|event| hypercall_db::EngineJournalEventInsert {
                event_topic: event.event_topic.clone(),
                event_key: event.event_key.clone(),
                event_data: event.event_data.clone(),
                l2_sequence: event.l2_sequence,
                event_type: event.event_type_enum.into(),
            })
            .collect(),
        outbox_appends: entry.outbox_appends.clone(),
        fill_side_effects: entry
            .fill_side_effects
            .iter()
            .map(|side_effect| hypercall_db::EngineJournalFillSideEffect {
                event_idx: side_effect.event_idx,
                trade_id: side_effect.trade_id,
                taker_ledger_delta: side_effect.taker_ledger_delta,
                maker_ledger_delta: side_effect.maker_ledger_delta,
                taker_premium_delta: side_effect.taker_premium_delta,
                maker_premium_delta: side_effect.maker_premium_delta,
                underlying_notional: side_effect.underlying_notional,
            })
            .collect(),
        cash_withdrawal_side_effect: entry.cash_withdrawal_side_effect.as_ref().map(
            |side_effect| hypercall_db::EngineJournalCashWithdrawalSideEffect {
                wallet: side_effect.wallet,
                request_id: side_effect.request_id.clone(),
                amount: side_effect.amount,
                balance_after: side_effect.balance_after,
                timestamp_ms: side_effect.timestamp_ms,
            },
        ),
        balance_updates: entry
            .balance_updates
            .iter()
            .cloned()
            .map(|update| hypercall_db::EngineJournalBalanceUpdate { update })
            .collect(),
    }
}

#[cfg(feature = "rsm-state")]
fn engine_command_identity_hashes(entries: &[JournalEntry]) -> Vec<[u8; 32]> {
    entries
        .iter()
        .filter_map(|entry| {
            entry
                .command_type_enum
                .is_some_and(rsm_state_supported_command)
                .then_some(entry.command_identity_hash)
        })
        .collect()
}

#[cfg(feature = "rsm-state")]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
struct RsmBatchRoots {
    state_root: [u8; 32],
    risk_root: [u8; 32],
    command_mmr_root: [u8; 32],
    obligation_mmr_root: [u8; 32],
    intent_mmr_root: [u8; 32],
    batch_root: [u8; 32],
}

#[cfg(feature = "rsm-state")]
impl From<hypercall_state_commitment::pipeline::BatchCommitment> for RsmBatchRoots {
    fn from(commitment: hypercall_state_commitment::pipeline::BatchCommitment) -> Self {
        Self {
            state_root: commitment.state_root.0,
            risk_root: commitment.risk_root.0,
            command_mmr_root: commitment.command_mmr_root.0,
            obligation_mmr_root: commitment.obligation_mmr_root.0,
            intent_mmr_root: commitment.intent_mmr_root.0,
            batch_root: commitment.batch_root,
        }
    }
}

#[cfg(feature = "rsm-state")]
fn rsm_batch_roots_from_wal(records: &[WalRecordWithOffset]) -> Option<RsmBatchRoots> {
    records.iter().find_map(|record| record.block_roots)
}

#[cfg(feature = "rsm-state")]
fn apply_recovered_rsm_commitment_blocks(
    records: &mut [WalRecordWithOffset],
    commitments: &mut ValidatorRsmStateCommitmentRuntime,
) -> anyhow::Result<()> {
    let mut index = 0;
    while index < records.len() {
        let Some(header) = records[index].block_header.as_ref() else {
            index += 1;
            continue;
        };

        let end_offset = records[index].end_offset;
        let block_end = records[index..]
            .iter()
            .position(|candidate| candidate.end_offset != end_offset)
            .map(|relative| index + relative)
            .unwrap_or(records.len());
        let block_records = &records[index..block_end];
        let expected_roots = rsm_batch_roots_from_wal(block_records);

        let next_version = commitments.next_batch_version();
        if header.batch_seq < next_version {
            index = block_end;
            continue;
        }
        if header.batch_seq > next_version {
            anyhow::bail!(
                "recovered RSM block version gap: WAL block {} but local commitment store next version is {}",
                header.batch_seq,
                next_version
            );
        }

        let entries: Vec<JournalEntry> = block_records
            .iter()
            .map(WalRecordWithOffset::to_db_entry)
            .collect();
        let identity_hashes = engine_command_identity_hashes(&entries);
        if identity_hashes.len() as u64 != header.command_count {
            anyhow::bail!(
                "recovered RSM block {} command count mismatch: header={} identities={}",
                header.batch_seq,
                header.command_count,
                identity_hashes.len()
            );
        }

        let delta = rsm_state_delta_from_entries(&entries, &identity_hashes, header.first_seq)?;
        let prepared = commitments.prepare_rsm_batch(&delta)?;
        let computed_roots: RsmBatchRoots = prepared.commitment.clone().into();
        match expected_roots {
            Some(expected_roots) if computed_roots != expected_roots => {
                anyhow::bail!(
                    "recovered RSM block {} root mismatch: WAL batch_root={} local batch_root={}",
                    header.batch_seq,
                    hex::encode(expected_roots.batch_root),
                    hex::encode(computed_roots.batch_root)
                );
            }
            Some(_) => {}
            None if computed_roots.batch_root != header.batch_root => {
                anyhow::bail!(
                    "recovered legacy RSM block {} root mismatch: WAL header batch_root={} local batch_root={}",
                    header.batch_seq,
                    hex::encode(header.batch_root),
                    hex::encode(computed_roots.batch_root)
                );
            }
            None => {
                records[index].block_roots = Some(computed_roots);
            }
        }
        commitments.apply_prepared_rsm_batch(prepared)?;

        index = block_end;
    }

    Ok(())
}

#[cfg(feature = "rsm-state")]
fn rsm_state_delta_from_entries(
    entries: &[JournalEntry],
    identity_hashes: &[[u8; 32]],
    first_seq: u64,
) -> anyhow::Result<StateDelta> {
    let engine_entries: Vec<&JournalEntry> = entries
        .iter()
        .filter(|entry| {
            entry
                .command_type_enum
                .is_some_and(rsm_state_supported_command)
        })
        .collect();
    let mut delta = rsm_state_delta_from_parts(
        engine_entries
            .iter()
            .filter_map(|entry| entry.command_type_enum),
        engine_entries.iter().flat_map(|entry| entry.events.iter()),
        identity_hashes,
        first_seq,
    )?;
    delta.global = Some(rsm_global_leaf(identity_hashes, first_seq));
    Ok(delta)
}

#[cfg(feature = "rsm-state")]
fn rsm_state_delta_from_parts<'a>(
    command_types: impl IntoIterator<Item = hypercall_db_diesel::engine_enums::CommandType>,
    events: impl IntoIterator<Item = &'a EventPayload>,
    identity_hashes: &[[u8; 32]],
    first_seq: u64,
) -> anyhow::Result<StateDelta> {
    for command_type in command_types {
        if !rsm_state_supported_command(command_type) {
            anyhow::bail!(
                "unsupported validator RSM state materializer command type: {}",
                command_type.as_str()
            );
        }
    }
    let mut delta = StateDelta::default();
    let _ = first_seq;
    for hash in identity_hashes {
        delta = append_command_delta(delta, *hash);
    }
    for event in events {
        delta = append_obligation_delta(delta, rsm_event_payload_hash(event));
    }
    for hash in identity_hashes {
        delta = append_intent_delta(delta, *hash);
    }
    Ok(delta)
}

#[cfg(feature = "rsm-state")]
fn append_command_delta(mut delta: StateDelta, command_hash: [u8; 32]) -> StateDelta {
    delta.commands.push(command_hash);
    delta
}

#[cfg(feature = "rsm-state")]
fn append_obligation_delta(mut delta: StateDelta, obligation_hash: [u8; 32]) -> StateDelta {
    delta.obligations.push(obligation_hash);
    delta
}

#[cfg(feature = "rsm-state")]
fn append_intent_delta(mut delta: StateDelta, intent_hash: [u8; 32]) -> StateDelta {
    delta.intents.push(intent_hash);
    delta
}

#[cfg(feature = "rsm-state")]
fn rsm_state_supported_command(
    command_type: hypercall_db_diesel::engine_enums::CommandType,
) -> bool {
    // First materializer slice: TickSnapshot only. It advances the command log
    // commitment but does not mutate order, position, ledger, market, oracle,
    // nonce, or risk state. Every state-mutating command fails closed until its
    // leaf materializer lands.
    matches!(
        command_type,
        hypercall_db_diesel::engine_enums::CommandType::TickSnapshot
    )
}

#[cfg(feature = "rsm-state")]
fn rsm_global_leaf(identity_hashes: &[[u8; 32]], first_seq: u64) -> GlobalLeaf {
    GlobalLeaf {
        next_order_id: 0,
        next_trade_id: 0,
        command_chain_root: hypercall_blocks::compute_commands_hash(identity_hashes),
        command_chain_seq: first_seq + identity_hashes.len() as u64,
    }
}

#[cfg(feature = "rsm-state")]
fn rsm_event_payload_hash(event: &EventPayload) -> [u8; 32] {
    let bytes = rmp_serde::to_vec_named(event)
        .expect("EventPayload serialization should not fail for RSM obligation root");
    rsm_hash_with_domain(b"hypercall:rsm:event-payload:v1", &bytes)
}

#[cfg(feature = "rsm-state")]
fn rsm_hash_with_domain(domain: &[u8], payload: &[u8]) -> [u8; 32] {
    use sha3::{Digest, Keccak256};

    let mut hasher = Keccak256::new();
    hasher.update(domain);
    hasher.update((payload.len() as u64).to_be_bytes());
    hasher.update(payload);
    hasher.finalize().into()
}

#[derive(Debug, Clone, Serialize, Deserialize)]
struct WalJournalRecord {
    request_uuid: hypercall_db_diesel::engine_enums::DbUuid,
    received_ts_ms: u64,
    command_data: Vec<u8>,
    response_data: Option<Vec<u8>>,
    order_id: Option<i64>,
    pre_digest: EngineStateDigest,
    post_digest: EngineStateDigest,
    duration_ms: u64,
    events: Vec<EventPayload>,
    outbox_appends: Vec<crate::directive_outbox::DirectiveOutboxAppend>,
    fill_side_effects: Vec<JournalFillSideEffect>,
    #[serde(default)]
    cash_withdrawal_side_effect: Option<JournalCashWithdrawalSideEffect>,
    #[serde(default)]
    balance_updates: Vec<hypercall_types::BalanceUpdate>,
    command_type_enum: Option<hypercall_db_diesel::engine_enums::CommandType>,
    #[cfg(feature = "rsm-state")]
    #[serde(default)]
    command_identity_hash: [u8; 32],
    #[cfg(feature = "rsm-state")]
    #[serde(default)]
    rsm_state_digest: Option<RsmEngineStateDigest>,
}

#[cfg(feature = "rsm-state")]
type WalBlockRecord = BlockLogHeader;
#[cfg(feature = "rsm-state")]
type WalBlockEntry = BlockLogEntry<WalJournalRecord>;

#[cfg(feature = "rsm-state")]
#[derive(Debug, Clone, Serialize, Deserialize)]
struct WalBlockRecordWithRoots {
    header: WalBlockRecord,
    roots: RsmBatchRoots,
}

#[cfg(feature = "rsm-state")]
#[derive(Debug, Clone, Serialize, Deserialize)]
struct WalDurableBlockEntry {
    header: WalBlockRecord,
    roots: RsmBatchRoots,
    commands: Vec<WalJournalRecord>,
}

impl WalJournalRecord {
    fn from_entry(entry: &JournalEntry) -> Self {
        Self {
            request_uuid: entry.request_uuid,
            received_ts_ms: entry.received_ts_ms,
            command_data: entry.command_data.clone(),
            response_data: entry.response_data.clone(),
            order_id: entry.order_id,
            pre_digest: entry.pre_digest.clone(),
            post_digest: entry.post_digest.clone(),
            duration_ms: entry.duration_ms,
            events: entry.events.clone(),
            outbox_appends: entry.outbox_appends.clone(),
            fill_side_effects: entry.fill_side_effects.clone(),
            cash_withdrawal_side_effect: entry.cash_withdrawal_side_effect.clone(),
            balance_updates: entry.balance_updates.clone(),
            command_type_enum: entry.command_type_enum,
            #[cfg(feature = "rsm-state")]
            command_identity_hash: entry.command_identity_hash,
            #[cfg(feature = "rsm-state")]
            rsm_state_digest: entry.rsm_state_digest.clone(),
        }
    }
}

#[derive(Debug, Clone)]
struct WalRecordWithOffset {
    record: WalJournalRecord,
    end_offset: u64,
    #[cfg(feature = "rsm-state")]
    block_header: Option<WalBlockRecord>,
    #[cfg(feature = "rsm-state")]
    block_roots: Option<RsmBatchRoots>,
}

impl WalRecordWithOffset {
    fn to_db_entry(&self) -> JournalEntry {
        JournalEntry {
            request_uuid: self.record.request_uuid,
            received_ts_ms: self.record.received_ts_ms,
            command_data: self.record.command_data.clone(),
            response_data: self.record.response_data.clone(),
            order_id: self.record.order_id,
            pre_digest: self.record.pre_digest.clone(),
            post_digest: self.record.post_digest.clone(),
            duration_ms: self.record.duration_ms,
            events: self.record.events.clone(),
            outbox_appends: self.record.outbox_appends.clone(),
            fill_side_effects: self.record.fill_side_effects.clone(),
            cash_withdrawal_side_effect: self.record.cash_withdrawal_side_effect.clone(),
            balance_updates: self.record.balance_updates.clone(),
            created_at: Instant::now(),
            commit_ack: None,
            command_type_enum: self.record.command_type_enum,
            #[cfg(feature = "rsm-state")]
            command_identity_hash: self.record.command_identity_hash,
            #[cfg(feature = "rsm-state")]
            rsm_state_digest: self.record.rsm_state_digest.clone(),
        }
    }
}

/// Returns the number of records for the next replication chunk.
///
/// `pending` is the ordered WAL tail and `max_batch_size` is the target command
/// cap. The returned length may exceed that cap only to keep records with the
/// same `end_offset` together, since they came from the same physical WAL
/// record and checkpoint advancement must stay atomic at that boundary.
fn next_replication_chunk_len(pending: &[WalRecordWithOffset], max_batch_size: usize) -> usize {
    if pending.is_empty() {
        return 0;
    }

    let mut chunk_len = max_batch_size.max(1).min(pending.len());
    while chunk_len < pending.len()
        && pending[chunk_len - 1].end_offset == pending[chunk_len].end_offset
    {
        chunk_len += 1;
    }
    chunk_len
}

fn should_defer_replay_record(
    current_batch_commands: usize,
    decoded_record_commands: usize,
    max_batch_size: usize,
) -> bool {
    current_batch_commands > 0
        && current_batch_commands + decoded_record_commands > max_batch_size.max(1)
}

struct WalState {
    wal_path: PathBuf,
    checkpoint_path: PathBuf,
    file: File,
    next_offset: u64,
    /// End offset of the last fallocate'd region. Pre-allocation avoids
    /// extent allocation stalls during sequential WAL writes on ext4/xfs.
    preallocated_end: u64,
}

impl WalState {
    /// Open a WAL at a specific path. Preferred for tests to avoid env var races.
    #[cfg(test)]
    fn open_at(wal_path: PathBuf) -> Result<Self, String> {
        if let Some(parent) = wal_path.parent() {
            if !parent.as_os_str().is_empty() {
                create_dir_all(parent).map_err(|e| {
                    format!("failed to create WAL directory {}: {}", parent.display(), e)
                })?;
            }
        }

        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        let file = OpenOptions::new()
            .create(true)
            .append(true)
            .read(true)
            .open(&wal_path)
            .map_err(|e| format!("failed to open WAL file {}: {}", wal_path.display(), e))?;

        let next_offset = file
            .metadata()
            .map_err(|e| format!("failed to stat WAL file {}: {}", wal_path.display(), e))?
            .len();

        Ok(Self {
            wal_path,
            checkpoint_path,
            file,
            next_offset,
            preallocated_end: next_offset,
        })
    }

    fn open(wal_path: PathBuf) -> Result<Self, String> {
        if let Some(parent) = wal_path.parent() {
            if !parent.as_os_str().is_empty() {
                create_dir_all(parent).map_err(|e| {
                    format!("failed to create WAL directory {}: {}", parent.display(), e)
                })?;

                Self::emit_disk_space_metrics(parent);
            }
        }

        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        #[cfg(unix)]
        let file = {
            use std::os::unix::fs::OpenOptionsExt;
            OpenOptions::new()
                .create(true)
                .append(true)
                .read(true)
                .mode(0o600)
                .open(&wal_path)
                .map_err(|e| format!("failed to open WAL file {}: {}", wal_path.display(), e))?
        };
        #[cfg(not(unix))]
        let file = OpenOptions::new()
            .create(true)
            .append(true)
            .read(true)
            .open(&wal_path)
            .map_err(|e| format!("failed to open WAL file {}: {}", wal_path.display(), e))?;

        let next_offset = file
            .metadata()
            .map_err(|e| format!("failed to stat WAL file {}: {}", wal_path.display(), e))?
            .len();

        // Punch holes for already-replicated WAL prefix on startup.
        // Reclaims disk from data written before hole punching was added.
        let cp = read_checkpoint(&checkpoint_path).unwrap_or_else(|e| {
            panic!(
                "CRITICAL_FAILURE: failed to read/parse checkpoint {}: {}",
                checkpoint_path.display(),
                e
            )
        });
        if cp.wal_offset > 0 {
            info!(
                "WAL startup: punching replicated prefix (0..{} bytes, file size={} bytes)",
                cp.wal_offset, next_offset
            );
            EngineJournalBatcher::punch_wal_hole(&wal_path, 0, cp.wal_offset);

            if let Some(parent) = wal_path.parent() {
                Self::emit_disk_space_metrics(parent);
            }
        }

        Ok(Self {
            wal_path,
            checkpoint_path,
            file,
            next_offset,
            preallocated_end: next_offset,
        })
    }

    /// Emit Prometheus gauges for WAL volume disk space.
    fn emit_disk_space_metrics(path: &Path) {
        #[cfg(unix)]
        {
            use std::ffi::CString;
            let c_path = match CString::new(path.to_string_lossy().as_bytes()) {
                Ok(p) => p,
                Err(_) => return,
            };
            unsafe {
                let mut stat: libc::statvfs = std::mem::zeroed();
                if libc::statvfs(c_path.as_ptr(), &mut stat) == 0 {
                    let total_bytes = stat.f_frsize as u64 * stat.f_blocks as u64;
                    let avail_bytes = stat.f_frsize as u64 * stat.f_bavail as u64;
                    let used_bytes =
                        total_bytes.saturating_sub(stat.f_frsize as u64 * stat.f_bfree as u64);
                    let pct = if total_bytes > 0 {
                        (used_bytes as f64 / total_bytes as f64) * 100.0
                    } else {
                        0.0
                    };

                    gauge!("ht_wal_volume_total_bytes").set(total_bytes as f64);
                    gauge!("ht_wal_volume_used_bytes").set(used_bytes as f64);
                    gauge!("ht_wal_volume_avail_bytes").set(avail_bytes as f64);
                    gauge!("ht_wal_volume_usage_pct").set(pct);
                }
            }
        }
    }

    /// Pre-allocate disk space for the WAL file in 64 MiB chunks using fallocate(2).
    ///
    /// Uses `FALLOC_FL_KEEP_SIZE` to avoid inflating the file size (replay reads
    /// up to the actual EOF). Non-fatal on failure (some filesystems like tmpfs
    /// don't support fallocate). No-op on non-Linux.
    fn ensure_preallocated(&mut self) -> u64 {
        if self.next_offset < self.preallocated_end {
            return 0;
        }

        #[cfg(target_os = "linux")]
        {
            let offset = self.preallocated_end as i64;
            let len = WAL_PREALLOC_CHUNK as i64;
            let ret = unsafe {
                libc::fallocate(
                    self.file.as_raw_fd(),
                    libc::FALLOC_FL_KEEP_SIZE,
                    offset,
                    len,
                )
            };
            if ret == 0 {
                self.preallocated_end += WAL_PREALLOC_CHUNK;
                WAL_PREALLOC_CHUNK
            } else {
                let errno = std::io::Error::last_os_error();
                warn!(
                    "fallocate failed (non-fatal, continuing without preallocation): {}",
                    errno
                );
                // Disk might be full — log space info for diagnosis
                if let Some(parent) = self.wal_path.parent() {
                    Self::emit_disk_space_metrics(parent);
                }
                0
            }
        }

        #[cfg(not(target_os = "linux"))]
        {
            0
        }
    }

    /// Serialize entries and write them to page cache via BufWriter.
    ///
    /// Does NOT call fsync — the caller (wal_writer_thread_main) is responsible
    /// for durability via io_uring FSYNC or sync_data().
    #[cfg_attr(feature = "rsm-state", allow(dead_code))]
    fn append_batch_no_sync(
        &mut self,
        entries: &[JournalEntry],
    ) -> Result<Vec<WalRecordWithOffset>, String> {
        if entries.is_empty() {
            return Ok(Vec::new());
        }

        // Ensure we have pre-allocated space to avoid extent allocation stalls.
        self.ensure_preallocated();

        let mut out = Vec::with_capacity(entries.len());

        let mut writer = BufWriter::new(&self.file);

        for entry in entries {
            let record = WalJournalRecord::from_entry(entry);
            let payload = rmp_serde::to_vec_named(&record).map_err(|e| {
                format!(
                    "failed to serialize WAL record for {}: {}",
                    entry.request_uuid, e
                )
            })?;

            write_frame(
                &mut writer,
                &payload,
                &mut self.next_offset,
                &format!("record for {}", entry.request_uuid),
            )?;

            out.push(WalRecordWithOffset {
                record,
                end_offset: self.next_offset,
                #[cfg(feature = "rsm-state")]
                block_header: None,
                #[cfg(feature = "rsm-state")]
                block_roots: None,
            });
        }

        writer
            .flush()
            .map_err(|e| format!("failed to flush WAL BufWriter: {}", e))?;

        Ok(out)
    }

    /// Append a self-contained validator RSM batch record.
    #[cfg(feature = "rsm-state")]
    fn append_block<T: Serialize>(&mut self, block: &T) -> Result<u64, String> {
        self.ensure_preallocated();

        let payload = rmp_serde::to_vec_named(block)
            .map_err(|e| format!("failed to serialize WAL block: {}", e))?;

        let mut writer = BufWriter::new(&self.file);
        write_frame(&mut writer, &payload, &mut self.next_offset, "block")?;
        writer
            .flush()
            .map_err(|e| format!("failed to flush WAL BufWriter: {}", e))?;

        Ok(self.next_offset)
    }

    /// Convenience method for tests: write + sync in one call.
    #[cfg(test)]
    fn append_batch(
        &mut self,
        entries: &[JournalEntry],
    ) -> Result<Vec<WalRecordWithOffset>, String> {
        let result = self.append_batch_no_sync(entries)?;
        self.file.sync_data().map_err(|e| {
            format!(
                "test sync_data failed for WAL {}: {}",
                self.wal_path.display(),
                e
            )
        })?;
        Ok(result)
    }

    #[cfg(test)]
    fn read_unreplicated_records(
        wal_path: &Path,
        checkpoint_path: &Path,
    ) -> Result<Vec<WalRecordWithOffset>, String> {
        if !wal_path.exists() {
            return Ok(Vec::new());
        }

        let checkpoint = read_checkpoint(checkpoint_path)?;
        let mut file = OpenOptions::new().read(true).open(wal_path).map_err(|e| {
            format!(
                "failed to open WAL {} for replay: {}",
                wal_path.display(),
                e
            )
        })?;

        let file_len = file
            .metadata()
            .map_err(|e| format!("failed to stat WAL {}: {}", wal_path.display(), e))?
            .len();

        if checkpoint.wal_offset > file_len {
            return Err(format!(
                "WAL checkpoint {} exceeds file length {} for {}",
                checkpoint.wal_offset,
                file_len,
                wal_path.display()
            ));
        }

        file.seek(SeekFrom::Start(checkpoint.wal_offset))
            .map_err(|e| format!("failed to seek WAL replay start: {}", e))?;

        let mut offset = checkpoint.wal_offset;
        let mut out = Vec::new();

        loop {
            let mut len_buf = [0u8; 4];
            match file.read_exact(&mut len_buf) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break,
                Err(e) => {
                    return Err(format!("failed reading WAL length prefix: {}", e));
                }
            }

            let payload_len = u32::from_le_bytes(len_buf) as usize;
            if payload_len == 0 {
                panic!(
                    "CRITICAL_FAILURE: WAL record length was zero at offset {} in {}. Data corruption detected.",
                    offset,
                    wal_path.display()
                );
            }
            offset += 4;

            // Guard against corrupt length prefix that could OOM the process.
            let remaining = file_len.saturating_sub(offset);
            if payload_len as u64 > remaining {
                warn!(
                    "WAL declared payload_len={} but only {} bytes remain at offset {} in {}, \
                     truncating tail and replaying prior durable records only",
                    payload_len,
                    remaining,
                    offset,
                    wal_path.display(),
                );
                break;
            }

            let mut crc_buf = [0u8; 4];
            match file.read_exact(&mut crc_buf) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    warn!(
                        "WAL ended with incomplete CRC header at offset {} in {} ({}), \
                         truncating tail and replaying prior durable records only",
                        offset,
                        wal_path.display(),
                        e
                    );
                    break;
                }
                Err(e) => return Err(format!("failed reading WAL crc prefix: {}", e)),
            }
            offset += 4;

            let expected_crc = u32::from_le_bytes(crc_buf);
            let mut payload = vec![0u8; payload_len];

            match file.read_exact(&mut payload) {
                Ok(()) => {}
                Err(e) if e.kind() == std::io::ErrorKind::UnexpectedEof => {
                    warn!(
                        "WAL ended with incomplete payload at offset {} in {} ({}), \
                         truncating tail and replaying prior durable records only",
                        offset,
                        wal_path.display(),
                        e
                    );
                    break;
                }
                Err(e) => return Err(format!("failed reading WAL payload: {}", e)),
            }
            offset += payload_len as u64;

            let actual_crc = WAL_CRC.checksum(&payload);
            if actual_crc != expected_crc {
                panic!(
                    "CRITICAL_FAILURE: WAL CRC mismatch at end_offset {} in {}: expected {}, got {}. Data corruption detected.",
                    offset,
                    wal_path.display(),
                    expected_crc,
                    actual_crc
                );
            }

            #[cfg(feature = "rsm-state")]
            if let Ok(block_entry) = rmp_serde::from_slice::<WalDurableBlockEntry>(&payload) {
                let header = block_entry.header;
                let roots = block_entry.roots;
                for (index, cmd) in block_entry.commands.into_iter().enumerate() {
                    out.push(WalRecordWithOffset {
                        record: cmd,
                        end_offset: offset,
                        block_header: (index == 0).then_some(header.clone()),
                        block_roots: (index == 0).then_some(roots),
                    });
                }
                continue;
            }

            #[cfg(feature = "rsm-state")]
            if rmp_serde::from_slice::<WalBlockEntry>(&payload).is_ok() {
                panic!(
                    "CRITICAL_FAILURE: legacy rootless RSM WAL block at end_offset {} is unsupported after durable root cutover",
                    offset
                );
            }

            if let Ok(record) = rmp_serde::from_slice::<WalJournalRecord>(&payload) {
                out.push(WalRecordWithOffset {
                    record,
                    end_offset: offset,
                    #[cfg(feature = "rsm-state")]
                    block_header: None,
                    #[cfg(feature = "rsm-state")]
                    block_roots: None,
                });
                continue;
            }

            #[cfg(feature = "rsm-state")]
            if rmp_serde::from_slice::<WalBlockRecord>(&payload).is_ok() {
                continue;
            }

            return Err(format!(
                "failed to deserialize WAL payload at end_offset {}",
                offset
            ));
        }

        Ok(out)
    }
}

/// Submit `IORING_OP_FSYNC` with `DATASYNC` flag via io_uring.
///
/// Only the fsync uses io_uring — writes go through regular BufWriter to page cache.
/// io_uring FSYNC flushes page cache to storage without blocking the thread on a
/// syscall (the kernel does the work asynchronously and we wait on the CQE).
#[cfg(target_os = "linux")]
fn wal_fsync_io_uring(ring: &mut io_uring::IoUring, file: &File) -> Result<(), String> {
    let fd = io_uring::types::Fd(file.as_raw_fd());
    let fsync_e = io_uring::opcode::Fsync::new(fd)
        .flags(io_uring::types::FsyncFlags::DATASYNC)
        .build()
        .user_data(0x42);

    unsafe {
        ring.submission()
            .push(&fsync_e)
            .map_err(|e| format!("io_uring submission push failed: {:?}", e))?;
    }

    ring.submit_and_wait(1)
        .map_err(|e| format!("io_uring submit_and_wait failed: {}", e))?;

    let mut cq = ring.completion();
    cq.sync();
    let cqe = cq
        .next()
        .ok_or_else(|| "io_uring completion queue empty after submit_and_wait".to_string())?;

    if cqe.result() < 0 {
        return Err(format!(
            "io_uring FSYNC failed with errno {}",
            -cqe.result()
        ));
    }

    Ok(())
}

/// Dedicated WAL writer OS thread.
///
/// Owns `WalState` exclusively (no Arc/Mutex). Receives batches from the batcher
/// task, writes to page cache, performs fsync (io_uring on Linux, sync_data elsewhere),
/// fires durability acks, and enqueues records for async replication.
fn wal_writer_thread_main(
    mut wal_state: WalState,
    rx: std::sync::mpsc::Receiver<WalWriterMessage>,
    replication_tx: mpsc::Sender<ReplicationMessage>,
    #[cfg(feature = "rsm-state")] mut rsm_state: RsmWalWriterState,
) {
    // Try to create io_uring ring. Falls back to sync_data() if unavailable
    // (e.g. seccomp-restricted containers, older kernels).
    #[cfg(target_os = "linux")]
    let mut ring: Option<io_uring::IoUring> = match io_uring::IoUring::new(8) {
        Ok(r) => {
            info!("WAL writer using io_uring for DATASYNC");
            Some(r)
        }
        Err(e) => {
            warn!(
                "io_uring unavailable ({}), WAL writer falling back to blocking fdatasync",
                e
            );
            None
        }
    };

    /// Perform fdatasync on the WAL file using the best available method.
    #[cfg(target_os = "linux")]
    fn do_fsync(ring: &mut Option<io_uring::IoUring>, file: &File, wal_path: &Path) {
        if let Some(ref mut r) = ring {
            if let Err(e) = wal_fsync_io_uring(r, file) {
                // io_uring FSYNC failed at runtime. Disable and fall back to
                // blocking fdatasync rather than crashing the engine.
                warn!(
                    "io_uring FSYNC failed ({}), disabling io_uring and falling back to fdatasync",
                    e
                );
                counter!("ht_journal_wal_io_uring_fallbacks").increment(1);
                *ring = None;
                // Fall through to sync_data below
            } else {
                return;
            }
        }
        if let Err(e) = file.sync_data() {
            if let Some(parent) = wal_path.parent() {
                WalState::emit_disk_space_metrics(parent);
            }
            panic!(
                "CRITICAL_FAILURE: fdatasync failed for WAL {}: {}. Engine must stop.",
                wal_path.display(),
                e
            );
        }
    }

    #[cfg(not(target_os = "linux"))]
    fn do_fsync(file: &File, wal_path: &Path) {
        if let Err(e) = file.sync_data() {
            if let Some(parent) = wal_path.parent() {
                WalState::emit_disk_space_metrics(parent);
            }
            panic!(
                "CRITICAL_FAILURE: fdatasync failed for WAL {}: {}. Engine must stop.",
                wal_path.display(),
                e
            );
        }
    }

    loop {
        match rx.recv() {
            Ok(WalWriterMessage::WriteBatch {
                entries,
                ack_senders,
                completion,
            }) => {
                let wal_start = Instant::now();
                let entry_count = entries.len();

                #[cfg(feature = "rsm-state")]
                let prepared_block = rsm_state.prepare_block(&entries);

                #[cfg(feature = "rsm-state")]
                let append_result = if let Some(prepared) = prepared_block.as_ref() {
                    let commands: Vec<WalJournalRecord> =
                        entries.iter().map(WalJournalRecord::from_entry).collect();
                    let block_entry = WalDurableBlockEntry {
                        header: prepared.block.header.clone(),
                        roots: prepared.block.roots,
                        commands,
                    };
                    wal_state.append_block(&block_entry).map(|end_offset| {
                        let header = block_entry.header.clone();
                        let roots = block_entry.roots;
                        block_entry
                            .commands
                            .into_iter()
                            .enumerate()
                            .map(|(index, cmd)| WalRecordWithOffset {
                                record: cmd,
                                end_offset,
                                block_header: (index == 0).then_some(header.clone()),
                                block_roots: (index == 0).then_some(roots),
                            })
                            .collect::<Vec<_>>()
                    })
                } else {
                    wal_state.append_batch_no_sync(&entries)
                };

                #[cfg(not(feature = "rsm-state"))]
                let append_result = wal_state.append_batch_no_sync(&entries);

                match append_result {
                    Ok(wal_records) => {
                        #[cfg(feature = "rsm-state")]
                        let mut wal_records = wal_records;

                        #[cfg(target_os = "linux")]
                        do_fsync(&mut ring, &wal_state.file, &wal_state.wal_path);
                        #[cfg(not(target_os = "linux"))]
                        do_fsync(&wal_state.file, &wal_state.wal_path);

                        #[cfg(feature = "rsm-state")]
                        if let Some(prepared) = prepared_block {
                            let applied_block = rsm_state.apply_prepared_block(prepared);
                            for record in wal_records.iter_mut() {
                                if record.block_header.is_some() {
                                    record.block_header = Some(applied_block.header.clone());
                                    record.block_roots = Some(applied_block.roots);
                                    break;
                                }
                            }
                        }

                        histogram!("ht_journal_wal_append_seconds")
                            .record(wal_start.elapsed().as_secs_f64());
                        counter!("ht_journal_wal_batches_total").increment(1);
                        counter!("ht_journal_wal_entries_total").increment(entry_count as u64);
                        gauge!("ht_wal_file_size_bytes").set(wal_state.next_offset as f64);

                        // Update volume disk space gauges on every write
                        if let Some(parent) = wal_state.wal_path.parent() {
                            WalState::emit_disk_space_metrics(parent);
                        }

                        // Durability boundary reached: entries are on disk. Fire acks.
                        for tx in ack_senders.into_iter().flatten() {
                            let _ = tx.send(());
                        }

                        // Enqueue for async PostgreSQL replication.
                        if let Err(e) =
                            replication_tx.blocking_send(ReplicationMessage::Entries(wal_records))
                        {
                            // The WAL writer is shut down before the replication
                            // channel during graceful shutdown, so if we reach
                            // here the replication task has crashed unexpectedly.
                            error!(
                                "CRITICAL_FAILURE: Replication channel closed unexpectedly \
                                 (replication task likely crashed): {}. Aborting process.",
                                e
                            );
                            std::process::abort();
                        }

                        // Signal completion if requested (Flush callers await this).
                        if let Some(tx) = completion {
                            let _ = tx.send(());
                        }
                    }
                    Err(e) => {
                        if let Some(parent) = wal_state.wal_path.parent() {
                            WalState::emit_disk_space_metrics(parent);
                        }
                        panic!(
                            "CRITICAL_FAILURE: WAL append failed: {}. WAL size={} bytes. Engine must stop.",
                            e, wal_state.next_offset
                        );
                    }
                }
            }
            Ok(WalWriterMessage::FlushBarrier { completion }) => {
                // Barrier processed in FIFO order — all prior WriteBatches
                // have already been fsynced by the time we get here.
                let _ = completion.send(());
            }
            Ok(WalWriterMessage::Shutdown) | Err(_) => {
                info!("WAL writer thread shutting down");
                break;
            }
        }
    }
}

/// Shared sender that can be cloned and used from the engine.
pub type SharedJournalBatchSender = Arc<JournalBatchSender>;

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

    #[test]
    fn test_config_defaults() {
        let config = JournalBatcherConfig::default();
        assert_eq!(config.channel_capacity, DEFAULT_CHANNEL_CAPACITY);
        assert_eq!(config.max_batch_size, 100);
        assert_eq!(config.flush_interval_ms, 10);
    }

    #[test]
    fn engine_journal_insert_projection_preserves_transactional_side_effect_inputs() {
        let request_uuid = hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4());
        let wallet = hypercall_types::WalletAddress::from([1u8; 20]);
        let account = hypercall_types::WalletAddress::from([2u8; 20]);
        let signer = hypercall_types::WalletAddress::from([3u8; 20]);
        let destination = hypercall_types::WalletAddress::from([4u8; 20]);

        let outbox_append = hypercall_db::DirectiveOutboxAppend::needs_rsm_signature(
            "directive-1".to_string(),
            hypercall_types::directives::ActionKey::SystemWithdrawToken,
            wallet,
            account,
            signer,
            17,
            "idem-1".to_string(),
            vec![10, 11, 12],
            1234,
            Some(5678),
        );
        let fill_accounting = hypercall_types::FillAccounting {
            trade_id: 42,
            taker_realized_pnl: Decimal::new(-125, 2),
            maker_realized_pnl: Decimal::new(125, 2),
            taker_premium_delta: Decimal::new(-7, 2),
            maker_premium_delta: Decimal::new(7, 2),
            taker_net_cash_delta: Decimal::new(-132, 2),
            maker_net_cash_delta: Decimal::new(132, 2),
        };
        let fill_side_effect = JournalFillSideEffect::from_fill_accounting(0, &fill_accounting);
        let cash_withdrawal_side_effect = JournalCashWithdrawalSideEffect {
            wallet: destination,
            request_id: "cash-withdrawal-1".to_string(),
            amount: Decimal::new(2500, 2),
            balance_after: Decimal::new(7500, 2),
            timestamp_ms: 9999,
        };
        let balance_update = hypercall_types::BalanceUpdate {
            balance_update_seq: 7,
            wallet,
            delta: Decimal::new(2500, 2),
            balance_after: Decimal::new(7500, 2),
            reason: hypercall_types::BalanceUpdateReason::Deposit,
            reference_id: Some("deposit-1".to_string()),
            source_command_id: None,
            timestamp_ms: 9999,
        };
        let entry = JournalEntry {
            request_uuid,
            received_ts_ms: 1000,
            command_data: vec![1, 2, 3],
            response_data: Some(vec![4, 5]),
            order_id: Some(9),
            pre_digest: EngineStateDigest {
                next_order_id: 10,
                next_trade_id: 20,
                l2_seq: 30,
                symbols: Default::default(),
            },
            post_digest: EngineStateDigest {
                next_order_id: 11,
                next_trade_id: 21,
                l2_seq: 31,
                symbols: Default::default(),
            },
            duration_ms: 6,
            events: vec![EventPayload {
                event_topic: "order-updates".to_string(),
                event_key: Some("order-9".to_string()),
                event_data: vec![0, 1, 2],
                l2_sequence: Some(44),
                event_type_enum: hypercall_db_diesel::engine_enums::EventType::OrderUpdate,
            }],
            outbox_appends: vec![outbox_append.clone()],
            fill_side_effects: vec![fill_side_effect.clone()],
            cash_withdrawal_side_effect: Some(cash_withdrawal_side_effect.clone()),
            balance_updates: vec![balance_update.clone()],
            created_at: Instant::now(),
            commit_ack: None,
            command_type_enum: Some(hypercall_db_diesel::engine_enums::CommandType::CreateOrder),
            #[cfg(feature = "rsm-state")]
            command_identity_hash: [5u8; 32],
            #[cfg(feature = "rsm-state")]
            rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
        };

        let projected = engine_journal_entry_insert(&entry);

        assert_eq!(projected.request_uuid, request_uuid.0);
        assert_eq!(projected.received_ts_ms, entry.received_ts_ms);
        assert_eq!(projected.command_data, entry.command_data);
        assert_eq!(projected.response_data, entry.response_data);
        assert_eq!(projected.order_id, entry.order_id);
        assert_eq!(projected.command_type.as_deref(), Some("CreateOrder"));
        assert_eq!(projected.duration_ms, entry.duration_ms);
        assert!(!projected.pre_digest_data.is_empty());
        assert!(!projected.post_digest_data.is_empty());
        assert_ne!(projected.pre_digest_data, projected.post_digest_data);

        assert_eq!(projected.events.len(), 1);
        assert_eq!(projected.events[0].event_topic, "order-updates");
        assert_eq!(projected.events[0].event_key.as_deref(), Some("order-9"));
        assert_eq!(projected.events[0].event_data, vec![0, 1, 2]);
        assert_eq!(projected.events[0].l2_sequence, Some(44));
        assert_eq!(
            projected.events[0].event_type,
            hypercall_db::EventType::OrderUpdate
        );

        assert_eq!(projected.outbox_appends.len(), 1);
        assert_eq!(
            projected.outbox_appends[0].directive_id,
            outbox_append.directive_id
        );
        assert_eq!(projected.outbox_appends[0].nonce, outbox_append.nonce);
        assert_eq!(
            projected.outbox_appends[0].action_key,
            hypercall_types::directives::ActionKey::SystemWithdrawToken
        );

        assert_eq!(projected.fill_side_effects.len(), 1);
        assert_eq!(
            projected.fill_side_effects[0].event_idx,
            fill_side_effect.event_idx
        );
        assert_eq!(
            projected.fill_side_effects[0].trade_id,
            fill_side_effect.trade_id
        );
        assert_eq!(
            projected.fill_side_effects[0].taker_ledger_delta,
            fill_side_effect.taker_ledger_delta
        );
        assert_eq!(
            projected.fill_side_effects[0].maker_premium_delta,
            fill_side_effect.maker_premium_delta
        );

        let projected_cash = projected
            .cash_withdrawal_side_effect
            .expect("cash withdrawal side effect should be projected");
        assert_eq!(projected_cash.wallet, cash_withdrawal_side_effect.wallet);
        assert_eq!(
            projected_cash.request_id,
            cash_withdrawal_side_effect.request_id
        );
        assert_eq!(projected_cash.amount, cash_withdrawal_side_effect.amount);
        assert_eq!(
            projected_cash.balance_after,
            cash_withdrawal_side_effect.balance_after
        );
        assert_eq!(
            projected_cash.timestamp_ms,
            cash_withdrawal_side_effect.timestamp_ms
        );

        assert_eq!(projected.balance_updates.len(), 1);
        assert_eq!(projected.balance_updates[0].update, balance_update);
    }

    #[test]
    fn wal_roundtrip_single_record() {
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");

        let mut wal = WalState::open_at(wal_path).expect("open wal");
        let request_uuid = hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4());
        let balance_update = hypercall_types::BalanceUpdate {
            balance_update_seq: 3,
            wallet: hypercall_types::WalletAddress::from([9u8; 20]),
            delta: Decimal::new(-125, 2),
            balance_after: Decimal::new(875, 2),
            reason: hypercall_types::BalanceUpdateReason::Withdrawal,
            reference_id: Some("withdrawal-1".to_string()),
            source_command_id: None,
            timestamp_ms: 321,
        };

        let entry = JournalEntry {
            request_uuid,
            received_ts_ms: 123,
            command_data: vec![1, 2, 3],
            response_data: Some(vec![4, 5, 6]),
            order_id: Some(42),
            pre_digest: EngineStateDigest::default(),
            post_digest: EngineStateDigest::default(),
            duration_ms: 7,
            events: vec![EventPayload {
                event_type_enum: hypercall_db_diesel::engine_enums::EventType::OrderUpdate,
                event_topic: "order-updates".to_string(),
                event_key: Some("k".to_string()),
                event_data: vec![1, 2, 3],
                l2_sequence: Some(1),
            }],
            outbox_appends: vec![],
            fill_side_effects: vec![],
            cash_withdrawal_side_effect: None,
            balance_updates: vec![balance_update.clone()],
            created_at: Instant::now(),
            commit_ack: None,
            command_type_enum: Some(hypercall_db_diesel::engine_enums::CommandType::CreateOrder),
            #[cfg(feature = "rsm-state")]
            command_identity_hash: [0u8; 32],
            #[cfg(feature = "rsm-state")]
            rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
        };

        let offsets = wal.append_batch(&[entry]).expect("append wal batch");
        assert_eq!(offsets.len(), 1);

        write_checkpoint(
            &wal.checkpoint_path,
            WalCheckpointMetadata {
                wal_offset: 0,
                last_command_id: 0,
                last_l2_seq: 0,
            },
        )
        .expect("write checkpoint");
        let replayed = WalState::read_unreplicated_records(&wal.wal_path, &wal.checkpoint_path)
            .expect("replay");
        assert_eq!(replayed.len(), 1);
        assert_eq!(replayed[0].record.request_uuid, request_uuid);
        assert_eq!(
            replayed[0].record.balance_updates,
            vec![balance_update.clone()]
        );
        assert_eq!(replayed[0].to_db_entry().balance_updates[0], balance_update);
    }

    #[test]
    fn wal_checkpoint_prevents_double_replay() {
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");

        let mut wal = WalState::open_at(wal_path).expect("open wal");

        // Write 3 records
        let entries: Vec<JournalEntry> = (0..3)
            .map(|i| JournalEntry {
                request_uuid: hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4()),
                received_ts_ms: 100 + i,
                command_data: vec![i as u8],
                response_data: None,
                order_id: Some(i as i64),
                pre_digest: EngineStateDigest::default(),
                post_digest: EngineStateDigest::default(),
                duration_ms: 1,
                events: vec![],
                outbox_appends: vec![],
                fill_side_effects: vec![],
                cash_withdrawal_side_effect: None,
                balance_updates: Vec::new(),
                created_at: Instant::now(),
                commit_ack: None,
                command_type_enum: Some(
                    hypercall_db_diesel::engine_enums::CommandType::CreateOrder,
                ),
                #[cfg(feature = "rsm-state")]
                command_identity_hash: [0u8; 32],
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
            })
            .collect();

        let offsets = wal.append_batch(&entries).expect("append batch");
        assert_eq!(offsets.len(), 3);

        // No checkpoint: all 3 records should be returned
        let replayed = WalState::read_unreplicated_records(&wal.wal_path, &wal.checkpoint_path)
            .expect("replay all");
        assert_eq!(replayed.len(), 3);

        // Write checkpoint at end of last record
        let last_offset = offsets.last().unwrap().end_offset;
        write_checkpoint(
            &wal.checkpoint_path,
            WalCheckpointMetadata {
                wal_offset: last_offset,
                last_command_id: 0,
                last_l2_seq: 0,
            },
        )
        .expect("write checkpoint");

        // After checkpoint: 0 records should be returned (all already replicated)
        let replayed = WalState::read_unreplicated_records(&wal.wal_path, &wal.checkpoint_path)
            .expect("replay after checkpoint");
        assert_eq!(replayed.len(), 0, "Checkpoint should prevent double replay");

        // Partial checkpoint: checkpoint at end of record 1
        let partial_offset = offsets[0].end_offset;
        write_checkpoint(
            &wal.checkpoint_path,
            WalCheckpointMetadata {
                wal_offset: partial_offset,
                last_command_id: 0,
                last_l2_seq: 0,
            },
        )
        .expect("write partial checkpoint");

        let replayed = WalState::read_unreplicated_records(&wal.wal_path, &wal.checkpoint_path)
            .expect("replay after partial checkpoint");
        assert_eq!(
            replayed.len(),
            2,
            "Partial checkpoint should replay only unreplicated records"
        );
    }

    #[test]
    #[should_panic(expected = "CRITICAL_FAILURE: WAL record length was zero")]
    fn wal_replay_panics_on_zero_length_record() {
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        std::fs::write(&wal_path, 0u32.to_le_bytes()).expect("write malformed wal");
        WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("zero-length record must panic before returning");
    }

    #[test]
    #[should_panic(expected = "CRITICAL_FAILURE: WAL CRC mismatch")]
    fn wal_replay_panics_on_crc_mismatch() {
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        let mut bytes = Vec::new();
        bytes.extend_from_slice(&1u32.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.push(0xAB);
        std::fs::write(&wal_path, bytes).expect("write malformed wal");

        WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("crc mismatch must panic before returning");
    }

    #[test]
    fn wal_preallocation_extends_on_demand() {
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");

        let mut wal = WalState::open_at(wal_path).expect("open wal");

        // Initially preallocated_end == next_offset == 0
        assert_eq!(wal.preallocated_end, 0);

        // ensure_preallocated should attempt to extend (may be no-op on non-Linux/tmpfs)
        let allocated = wal.ensure_preallocated();

        #[cfg(target_os = "linux")]
        {
            // On Linux with a real filesystem, fallocate should succeed.
            // On tmpfs (CI), it may fail gracefully and return 0.
            // Either way, no panic.
            let _ = allocated;
        }

        #[cfg(not(target_os = "linux"))]
        {
            assert_eq!(allocated, 0, "Non-Linux should no-op");
        }
    }

    #[cfg(not(feature = "rsm-state"))]
    #[test]
    fn wal_writer_thread_shutdown() {
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");

        let wal_state = WalState::open_at(wal_path).expect("open wal");
        let (wal_writer_tx, wal_writer_rx) = std::sync::mpsc::sync_channel(16);
        let (replication_tx, mut replication_rx) = mpsc::channel(16);

        let handle = std::thread::Builder::new()
            .name("test-wal-writer".to_string())
            .spawn(move || {
                wal_writer_thread_main(
                    wal_state,
                    wal_writer_rx,
                    replication_tx,
                    #[cfg(feature = "rsm-state")]
                    test_rsm_writer_state(),
                );
            })
            .expect("spawn test wal-writer thread");

        // Send a batch
        let entry = JournalEntry {
            request_uuid: hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4()),
            received_ts_ms: 1,
            command_data: vec![1],
            response_data: None,
            order_id: None,
            pre_digest: EngineStateDigest::default(),
            post_digest: EngineStateDigest::default(),
            duration_ms: 0,
            events: vec![],
            outbox_appends: vec![],
            fill_side_effects: vec![],
            cash_withdrawal_side_effect: None,
            balance_updates: Vec::new(),
            created_at: Instant::now(),
            commit_ack: None,
            command_type_enum: Some(hypercall_db_diesel::engine_enums::CommandType::CreateOrder),
            #[cfg(feature = "rsm-state")]
            command_identity_hash: [0u8; 32],
            #[cfg(feature = "rsm-state")]
            rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
        };

        let (ack_tx, ack_rx) = oneshot::channel();
        let (complete_tx, complete_rx) = oneshot::channel();

        wal_writer_tx
            .send(WalWriterMessage::WriteBatch {
                entries: vec![entry],
                ack_senders: vec![Some(ack_tx)],
                completion: Some(complete_tx),
            })
            .expect("send to wal-writer");

        // Wait for completion
        complete_rx.blocking_recv().expect("completion signal");
        // Ack should have been fired
        ack_rx.blocking_recv().expect("ack signal");

        // Verify replication message was sent
        match replication_rx.blocking_recv() {
            Some(ReplicationMessage::Entries(records)) => {
                assert_eq!(records.len(), 1);
            }
            other => panic!("Expected Entries, got {:?}", other.is_some()),
        }

        // Shutdown
        wal_writer_tx
            .send(WalWriterMessage::Shutdown)
            .expect("send shutdown");
        handle.join().expect("wal-writer thread join");
    }

    // ---------------------------------------------------------------
    // WAL streaming recovery tests
    //
    // These exercise the WAL binary format (4-byte LE length, 4-byte LE CRC,
    // msgpack payload) via `read_unreplicated_records` which uses identical
    // parsing logic to `recover_wal_sync`.  The streaming variant additionally
    // batches records and writes them to PostgreSQL, but the I/O parsing is
    // shared.
    // ---------------------------------------------------------------

    /// Helper: build a valid WAL record byte sequence for a single JournalEntry.
    fn wal_bytes_for_entry(entry: &JournalEntry) -> Vec<u8> {
        let record = WalJournalRecord::from_entry(entry);
        let payload = rmp_serde::to_vec_named(&record).expect("serialize wal record");
        let payload_len = payload.len() as u32;
        let crc = WAL_CRC.checksum(&payload);

        let mut buf = Vec::with_capacity(8 + payload.len());
        buf.extend_from_slice(&payload_len.to_le_bytes());
        buf.extend_from_slice(&crc.to_le_bytes());
        buf.extend_from_slice(&payload);
        buf
    }

    /// Helper: build a minimal JournalEntry with a given index for differentiation.
    fn make_test_entry(index: u8) -> JournalEntry {
        JournalEntry {
            request_uuid: hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4()),
            received_ts_ms: 1000 + index as u64,
            command_data: vec![index],
            response_data: None,
            order_id: Some(index as i64),
            pre_digest: EngineStateDigest::default(),
            post_digest: EngineStateDigest::default(),
            duration_ms: 1,
            events: vec![],
            outbox_appends: vec![],
            fill_side_effects: vec![],
            cash_withdrawal_side_effect: None,
            balance_updates: Vec::new(),
            created_at: Instant::now(),
            commit_ack: None,
            command_type_enum: Some(hypercall_db_diesel::engine_enums::CommandType::CreateOrder),
            #[cfg(feature = "rsm-state")]
            command_identity_hash: [0u8; 32],
            #[cfg(feature = "rsm-state")]
            rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
        }
    }

    #[cfg(feature = "rsm-state")]
    fn make_test_event(index: u8) -> EventPayload {
        EventPayload {
            event_topic: "test.topic".to_string(),
            event_key: Some(format!("key-{index}")),
            event_data: vec![index, index.wrapping_add(1)],
            l2_sequence: Some(index as i64),
            event_type_enum: hypercall_db_diesel::engine_enums::EventType::OrderUpdate,
        }
    }

    #[cfg(feature = "rsm-state")]
    fn rsm_test_roots(entries: &[JournalEntry], identity_hashes: &[[u8; 32]]) -> RsmBatchRoots {
        let dir = tempfile::tempdir().expect("create tempdir");
        let store = crate::rsm::commitment_store::ValidatorRsmCommitmentStore::open(
            ValidatorRsmCommitmentStoreConfig {
                root_dir: dir.path().join("state"),
                prune_window: 100,
                pruning_enabled: false,
            },
        )
        .expect("open commitment store");
        let mut runtime = store.open_pipeline(0).expect("open pipeline");
        let delta = rsm_state_delta_from_entries(entries, identity_hashes, 0)
            .expect("materialize test delta");
        runtime.commit(&delta).expect("commit test delta").into()
    }

    #[cfg(feature = "rsm-state")]
    fn make_test_rsm_entry(index: u8) -> JournalEntry {
        let mut entry = make_test_entry(index);
        entry.command_type_enum =
            Some(hypercall_db_diesel::engine_enums::CommandType::TickSnapshot);
        entry
    }

    #[cfg(feature = "rsm-state")]
    fn make_test_rsm_block_records(
        batch_seq: u64,
        prev_block_hash: [u8; 32],
        first_seq: u64,
        entries: &[JournalEntry],
    ) -> Vec<WalRecordWithOffset> {
        let identity_hashes = engine_command_identity_hashes(entries);
        let roots = rsm_test_roots(entries, &identity_hashes);
        let header = WalBlockRecord::unsigned_from_identity_hashes(
            batch_seq,
            prev_block_hash,
            &identity_hashes,
            roots.batch_root,
            first_seq,
            1_700_000_000_000 + batch_seq,
            [0u8; 20],
        );

        entries
            .iter()
            .enumerate()
            .map(|(index, entry)| WalRecordWithOffset {
                record: WalJournalRecord::from_entry(entry),
                end_offset: 100 + batch_seq,
                block_header: (index == 0).then_some(header.clone()),
                block_roots: (index == 0).then_some(roots),
            })
            .collect()
    }

    #[test]
    fn wal_recovery_empty_wal_file() {
        // An empty WAL file (0 bytes) should recover zero records.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        // Create empty WAL file.
        std::fs::write(&wal_path, b"").expect("write empty wal");

        // No checkpoint file => defaults to offset 0.
        let records = WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("should recover empty wal");
        assert!(records.is_empty(), "empty WAL should yield zero records");
    }

    #[test]
    fn replication_chunk_extends_to_keep_physical_wal_block_atomic() {
        let pending: Vec<WalRecordWithOffset> = [(0, 100), (1, 100), (2, 100), (3, 200), (4, 200)]
            .into_iter()
            .map(|(index, end_offset)| WalRecordWithOffset {
                record: WalJournalRecord::from_entry(&make_test_entry(index)),
                end_offset,
                #[cfg(feature = "rsm-state")]
                block_header: None,
                #[cfg(feature = "rsm-state")]
                block_roots: None,
            })
            .collect();

        assert_eq!(next_replication_chunk_len(&pending, 2), 3);
        assert_eq!(next_replication_chunk_len(&pending[3..], 1), 2);
    }

    #[test]
    fn wal_replay_defers_next_physical_record_to_cap_command_batch() {
        assert!(!should_defer_replay_record(0, 100, 10));
        assert!(!should_defer_replay_record(5, 5, 10));
        assert!(should_defer_replay_record(5, 6, 10));
        assert!(should_defer_replay_record(1, 1, 0));
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    fn block_commitment_excludes_non_engine_journal_entries() {
        let mut engine_entry = make_test_rsm_entry(1);
        engine_entry.command_identity_hash = [1u8; 32];
        engine_entry.events.push(make_test_event(1));

        let mut passthrough_entry = make_test_entry(2);
        passthrough_entry.command_type_enum = None;
        passthrough_entry.command_identity_hash = [0u8; 32];
        passthrough_entry.events.push(make_test_event(2));

        let hashes = engine_command_identity_hashes(&[engine_entry, passthrough_entry]);

        assert_eq!(hashes, vec![[1u8; 32]]);
        assert_ne!(
            hypercall_blocks::compute_commands_hash(&hashes),
            hypercall_blocks::compute_commands_hash(&[[1u8; 32], [0u8; 32]])
        );
        let mut engine_only_entry = make_test_rsm_entry(1);
        engine_only_entry.command_identity_hash = [1u8; 32];
        engine_only_entry.events.push(make_test_event(1));

        let mut mixed_engine_entry = make_test_rsm_entry(1);
        mixed_engine_entry.command_identity_hash = [1u8; 32];
        mixed_engine_entry.events.push(make_test_event(1));

        let mut mixed_passthrough_entry = make_test_entry(2);
        mixed_passthrough_entry.command_type_enum = None;
        mixed_passthrough_entry.command_identity_hash = [0u8; 32];
        mixed_passthrough_entry.events.push(make_test_event(2));

        assert_eq!(
            rsm_test_roots(&[engine_only_entry], &hashes).obligation_mmr_root,
            rsm_test_roots(&[mixed_engine_entry, mixed_passthrough_entry], &hashes)
                .obligation_mmr_root
        );
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    fn block_commitment_excludes_unsupported_rsm_commands() {
        let mut supported_entry = make_test_rsm_entry(1);
        supported_entry.command_identity_hash = [1u8; 32];
        supported_entry.events.push(make_test_event(1));

        let mut agent_auth_entry = make_test_rsm_entry(2);
        agent_auth_entry.command_type_enum =
            Some(hypercall_db_diesel::engine_enums::CommandType::ApproveAgent);
        agent_auth_entry.command_identity_hash = [2u8; 32];

        let hashes = engine_command_identity_hashes(&[supported_entry, agent_auth_entry]);

        assert_eq!(hashes, vec![[1u8; 32]]);
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    fn rsm_batch_roots_are_deterministic_and_nonzero() {
        let mut entry = make_test_rsm_entry(1);
        entry.command_identity_hash = [9u8; 32];
        entry.events.push(make_test_event(1));

        let roots = rsm_test_roots(&[entry], &[[9u8; 32]]);
        let mut same_entry = make_test_rsm_entry(1);
        same_entry.command_identity_hash = [9u8; 32];
        same_entry.events.push(make_test_event(1));

        assert_ne!(roots.state_root, [0u8; 32]);
        assert_ne!(roots.command_mmr_root, [0u8; 32]);
        assert_ne!(roots.obligation_mmr_root, [0u8; 32]);
        assert_ne!(roots.intent_mmr_root, [0u8; 32]);
        assert_ne!(roots.batch_root, [0u8; 32]);
        assert_eq!(
            roots.batch_root,
            rsm_test_roots(&[same_entry], &[[9u8; 32]]).batch_root
        );
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    fn rsm_batch_roots_change_with_state_intents_and_obligations() {
        let mut entry = make_test_rsm_entry(1);
        entry.command_identity_hash = [9u8; 32];
        entry.events.push(make_test_event(1));
        let base = rsm_test_roots(&[entry], &[[9u8; 32]]);

        let mut changed_state = make_test_rsm_entry(1);
        changed_state.command_identity_hash = [9u8; 32];
        changed_state.events.push(make_test_event(1));
        assert_ne!(
            base.state_root,
            rsm_state_delta_from_entries(&[changed_state], &[[9u8; 32]], 7)
                .and_then(|delta| {
                    let dir = tempfile::tempdir().unwrap();
                    let store = crate::rsm::commitment_store::ValidatorRsmCommitmentStore::open(
                        ValidatorRsmCommitmentStoreConfig {
                            root_dir: dir.path().join("state"),
                            prune_window: 100,
                            pruning_enabled: false,
                        },
                    )?;
                    Ok(store.open_pipeline(0)?.commit(&delta)?.state_root.0)
                })
                .unwrap()
        );

        let mut same_entry = make_test_rsm_entry(1);
        same_entry.command_identity_hash = [9u8; 32];
        same_entry.events.push(make_test_event(1));
        assert_ne!(
            base.intent_mmr_root,
            rsm_test_roots(&[same_entry], &[[8u8; 32]]).intent_mmr_root
        );

        let mut changed_obligation = make_test_rsm_entry(1);
        changed_obligation.command_identity_hash = [9u8; 32];
        changed_obligation.events.push(make_test_event(1));
        changed_obligation.events[0].event_data.push(42);
        assert_ne!(
            base.obligation_mmr_root,
            rsm_test_roots(&[changed_obligation], &[[9u8; 32]]).obligation_mmr_root
        );
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    fn rsm_batch_roots_from_wal_tolerates_legacy_records_without_state_digest() {
        let mut entry = make_test_entry(1);
        entry.command_identity_hash = [1u8; 32];
        entry.rsm_state_digest = None;
        let record = WalRecordWithOffset {
            record: WalJournalRecord::from_entry(&entry),
            end_offset: 100,
            block_header: None,
            block_roots: None,
        };
        assert!(rsm_batch_roots_from_wal(&[record]).is_none());
    }

    #[test]
    fn wal_recovery_nonexistent_wal_file() {
        // If the WAL file doesn't exist, recovery is a no-op.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("nonexistent.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        let records = WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("missing WAL should be no-op");
        assert!(records.is_empty());
    }

    #[test]
    fn wal_recovery_small_wal_few_records() {
        // Write a few records via WalState, then read them back and verify
        // that field contents roundtrip correctly.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");

        let mut wal = WalState::open_at(wal_path).expect("open wal");

        let entries: Vec<JournalEntry> = (0..5).map(make_test_entry).collect();
        let request_uuids: Vec<_> = entries.iter().map(|e| e.request_uuid).collect();

        wal.append_batch(&entries).expect("append batch");

        // Checkpoint at 0 => all 5 records should be recovered.
        write_checkpoint(
            &wal.checkpoint_path,
            WalCheckpointMetadata {
                wal_offset: 0,
                last_command_id: 0,
                last_l2_seq: 0,
            },
        )
        .expect("write checkpoint");

        let records = WalState::read_unreplicated_records(&wal.wal_path, &wal.checkpoint_path)
            .expect("read records");
        assert_eq!(records.len(), 5);

        for (i, record) in records.iter().enumerate() {
            assert_eq!(
                record.record.request_uuid, request_uuids[i],
                "request_uuid mismatch at index {}",
                i
            );
            assert_eq!(record.record.command_data, vec![i as u8]);
            assert_eq!(record.record.received_ts_ms, 1000 + i as u64);
            assert_eq!(record.record.order_id, Some(i as i64));
        }

        // Verify offsets are monotonically increasing.
        for i in 1..records.len() {
            assert!(
                records[i].end_offset > records[i - 1].end_offset,
                "offsets must increase monotonically"
            );
        }
    }

    #[test]
    fn wal_recovery_resumes_from_checkpoint() {
        // Write 5 records, checkpoint after record 2, then recover only
        // the remaining 3.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");

        let mut wal = WalState::open_at(wal_path).expect("open wal");

        let entries: Vec<JournalEntry> = (0..5).map(make_test_entry).collect();
        let request_uuids: Vec<_> = entries.iter().map(|e| e.request_uuid).collect();

        let offsets = wal.append_batch(&entries).expect("append batch");
        assert_eq!(offsets.len(), 5);

        // Checkpoint after record index 1 (i.e. 2 records replicated).
        let checkpoint_offset = offsets[1].end_offset;
        write_checkpoint(
            &wal.checkpoint_path,
            WalCheckpointMetadata {
                wal_offset: checkpoint_offset,
                last_command_id: 0,
                last_l2_seq: 0,
            },
        )
        .expect("write checkpoint");

        let records = WalState::read_unreplicated_records(&wal.wal_path, &wal.checkpoint_path)
            .expect("read records after checkpoint");
        assert_eq!(records.len(), 3, "should recover only records 2,3,4");

        // Records should be the last 3 entries.
        for (i, record) in records.iter().enumerate() {
            let entry_index = i + 2;
            assert_eq!(
                record.record.request_uuid, request_uuids[entry_index],
                "request_uuid mismatch at recovered index {}",
                i
            );
        }
    }

    #[test]
    fn wal_recovery_truncated_after_length_prefix() {
        // Simulate a crash that left a partial record: length prefix is
        // written but CRC and payload are missing.  Recovery should warn
        // and return the records that were complete before the truncation.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        // Write one complete record, then append a truncated length prefix.
        let entry = make_test_entry(0);
        let mut wal_data = wal_bytes_for_entry(&entry);
        let complete_record_len = wal_data.len();

        // Append partial record: just the 4-byte length prefix, no CRC or payload.
        wal_data.extend_from_slice(&42u32.to_le_bytes());

        std::fs::write(&wal_path, &wal_data).expect("write wal with truncated tail");

        // No checkpoint => start from 0.
        let records = WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("should recover despite truncated tail");
        assert_eq!(records.len(), 1, "should recover the one complete record");
        assert_eq!(records[0].end_offset, complete_record_len as u64);
    }

    #[test]
    fn wal_recovery_truncated_after_crc() {
        // Simulate a crash that wrote the length + CRC but the payload
        // is incomplete.  Recovery should still return prior complete records.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        // Write two complete records, then append a truncated third.
        let entries: Vec<JournalEntry> = (0..2).map(make_test_entry).collect();
        let mut wal_data = Vec::new();
        for entry in &entries {
            wal_data.extend(wal_bytes_for_entry(entry));
        }

        // Truncated third record: length + CRC but only 1 byte of payload
        // (claims payload is 100 bytes).
        wal_data.extend_from_slice(&100u32.to_le_bytes()); // length = 100
        wal_data.extend_from_slice(&0u32.to_le_bytes()); // CRC (won't matter)
        wal_data.push(0xFF); // only 1 of 100 payload bytes

        std::fs::write(&wal_path, &wal_data).expect("write wal with truncated payload");

        let records = WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("should recover despite truncated payload");
        assert_eq!(
            records.len(),
            2,
            "should recover both complete records before the truncation"
        );
    }

    #[test]
    fn wal_recovery_truncated_mid_length_prefix() {
        // WAL ends with only 2 of 4 bytes of the next record's length prefix.
        // The read_exact for length should hit UnexpectedEof and stop cleanly.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        let entry = make_test_entry(0);
        let mut wal_data = wal_bytes_for_entry(&entry);

        // Append only 2 bytes of a length prefix.
        wal_data.extend_from_slice(&[0x01, 0x00]);

        std::fs::write(&wal_path, &wal_data).expect("write wal with partial length prefix");

        let records = WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("should recover despite partial length prefix");
        assert_eq!(records.len(), 1, "should recover the one complete record");
    }

    #[test]
    fn wal_recovery_corrupt_giant_payload_len() {
        // A corrupt length prefix that claims a multi-GB payload should not OOM
        // the process.  The payload_len guard should detect that the declared
        // length exceeds the remaining file bytes and stop replay.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        // Write one complete record first.
        let entry = make_test_entry(0);
        let mut wal_data = wal_bytes_for_entry(&entry);
        let complete_len = wal_data.len();

        // Append a record with a huge payload_len (2 GB) but only 16 bytes of
        // actual data.  Without the guard this would attempt a 2 GB allocation.
        wal_data.extend_from_slice(&0x8000_0000u32.to_le_bytes()); // 2 GB
        wal_data.extend_from_slice(&0u32.to_le_bytes()); // CRC placeholder
        wal_data.extend_from_slice(&[0xAB; 16]); // 16 bytes of junk

        std::fs::write(&wal_path, &wal_data).expect("write wal with corrupt length");

        let records = WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("should recover despite corrupt length prefix");
        assert_eq!(
            records.len(),
            1,
            "should recover the one complete record before the corrupt length"
        );
        assert_eq!(records[0].end_offset, complete_len as u64);
    }

    #[test]
    fn wal_recovery_payload_len_exactly_exceeds_remaining() {
        // Length prefix claims just a few more bytes than remain in the file.
        // This tests the boundary of the payload_len guard.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        let entry = make_test_entry(0);
        let mut wal_data = wal_bytes_for_entry(&entry);

        // Append a record whose payload_len (200) exceeds remaining bytes.
        // After the 4-byte length prefix, only 0 bytes of CRC+payload remain
        // because we write nothing else.
        wal_data.extend_from_slice(&200u32.to_le_bytes());

        std::fs::write(&wal_path, &wal_data).expect("write wal");

        let records = WalState::read_unreplicated_records(&wal_path, &checkpoint_path)
            .expect("should recover despite oversized payload_len");
        assert_eq!(records.len(), 1);
    }

    #[test]
    fn wal_recovery_manual_bytes_match_wal_state_writer() {
        // Verify that manually-constructed WAL bytes (as used in truncation
        // tests) produce the same records as WalState::append_batch.
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");

        let entry = make_test_entry(42);
        let manual_bytes = wal_bytes_for_entry(&entry);

        // Write the same entry via WalState.
        let mut wal = WalState::open_at(wal_path.clone()).expect("open wal");
        wal.append_batch(&[entry]).expect("append batch");
        let wal_state_bytes = std::fs::read(&wal_path).expect("read wal file");

        assert_eq!(
            manual_bytes, wal_state_bytes,
            "Manual WAL byte construction must match WalState writer output"
        );
    }

    // ---------------------------------------------------------------
    // WAL block entry serialization tests
    // ---------------------------------------------------------------

    #[cfg(feature = "rsm-state")]
    #[test]
    fn test_wal_block_entry_roundtrip() {
        // Create a WalBlockEntry with a header and 3 commands, serialize,
        // deserialize, and verify all fields match.
        let commands: Vec<WalJournalRecord> = (0..3)
            .map(|i| {
                let entry = make_test_entry(i);
                WalJournalRecord::from_entry(&entry)
            })
            .collect();

        let header = WalBlockRecord {
            batch_seq: 0,
            prev_block_hash: [0u8; 32],
            commands_hash: [1u8; 32],
            batch_root: [0u8; 32],
            command_count: 3,
            first_seq: 0,
            timestamp: 1700000000000,
            signer: [0u8; 20],
            signature: vec![0u8; 65],
        };

        let block_entry = WalBlockEntry {
            header: header.clone(),
            commands: commands.clone(),
        };

        let serialized = rmp_serde::to_vec_named(&block_entry).expect("serialize WalBlockEntry");
        let deserialized: WalBlockEntry =
            rmp_serde::from_slice(&serialized).expect("deserialize WalBlockEntry");

        // Verify header fields
        assert_eq!(deserialized.header.batch_seq, header.batch_seq);
        assert_eq!(deserialized.header.prev_block_hash, header.prev_block_hash);
        assert_eq!(deserialized.header.commands_hash, header.commands_hash);
        assert_eq!(deserialized.header.batch_root, header.batch_root);
        assert_eq!(deserialized.header.command_count, header.command_count);
        assert_eq!(deserialized.header.first_seq, header.first_seq);
        assert_eq!(deserialized.header.timestamp, header.timestamp);
        assert_eq!(deserialized.header.signer, header.signer);
        assert_eq!(deserialized.header.signature, header.signature);

        // Verify commands
        assert_eq!(deserialized.commands.len(), 3);
        for (i, cmd) in deserialized.commands.iter().enumerate() {
            assert_eq!(cmd.command_data, commands[i].command_data);
            assert_eq!(cmd.received_ts_ms, commands[i].received_ts_ms);
            assert_eq!(cmd.order_id, commands[i].order_id);
            assert_eq!(cmd.request_uuid, commands[i].request_uuid);
        }
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    fn test_wal_block_entry_backward_compat() {
        // Serialize an old-format WalJournalRecord, then verify:
        // 1. It fails to deserialize as WalBlockEntry (different structure)
        // 2. It succeeds as WalJournalRecord
        let entry = make_test_entry(42);
        let record = WalJournalRecord::from_entry(&entry);
        let serialized = rmp_serde::to_vec_named(&record).expect("serialize WalJournalRecord");

        // Should fail as WalBlockEntry (it has no 'header' / 'commands' fields)
        let block_attempt = rmp_serde::from_slice::<WalBlockEntry>(&serialized);
        assert!(
            block_attempt.is_err(),
            "Old-format WalJournalRecord must not deserialize as WalBlockEntry"
        );

        // Should succeed as WalJournalRecord
        let roundtripped: WalJournalRecord =
            rmp_serde::from_slice(&serialized).expect("deserialize WalJournalRecord");
        assert_eq!(roundtripped.request_uuid, record.request_uuid);
        assert_eq!(roundtripped.command_data, record.command_data);
        assert_eq!(roundtripped.received_ts_ms, record.received_ts_ms);
        assert_eq!(roundtripped.order_id, record.order_id);
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    #[should_panic(expected = "legacy rootless RSM WAL block")]
    fn wal_recovery_rejects_legacy_rootless_rsm_block() {
        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);
        let command = WalJournalRecord::from_entry(&make_test_entry(42));
        let block_entry = WalBlockEntry {
            header: WalBlockRecord {
                batch_seq: 0,
                prev_block_hash: [0u8; 32],
                commands_hash: [1u8; 32],
                batch_root: [2u8; 32],
                command_count: 1,
                first_seq: 0,
                timestamp: 1_700_000_000_000,
                signer: [0u8; 20],
                signature: vec![0u8; 65],
            },
            commands: vec![command],
        };
        let payload = rmp_serde::to_vec_named(&block_entry).expect("serialize legacy block");
        let mut file = File::create(&wal_path).expect("create wal");
        let mut offset = 0;
        write_frame(&mut file, &payload, &mut offset, "legacy block").expect("write frame");
        drop(file);

        let _ = WalState::read_unreplicated_records(&wal_path, &checkpoint_path);
    }

    #[cfg(feature = "rsm-state")]
    #[test]
    fn test_wal_block_entry_with_empty_commands() {
        // A block with 0 commands should serialize/deserialize correctly.
        let header = WalBlockRecord {
            batch_seq: 0,
            prev_block_hash: [0u8; 32],
            commands_hash: [0u8; 32],
            batch_root: [0u8; 32],
            command_count: 0,
            first_seq: 0,
            timestamp: 1700000000000,
            signer: [0u8; 20],
            signature: vec![0u8; 65],
        };

        let block_entry = WalBlockEntry {
            header: header.clone(),
            commands: vec![],
        };

        let serialized = rmp_serde::to_vec_named(&block_entry).expect("serialize empty block");
        let deserialized: WalBlockEntry =
            rmp_serde::from_slice(&serialized).expect("deserialize empty block");

        assert_eq!(deserialized.header.batch_seq, 0);
        assert_eq!(deserialized.header.command_count, 0);
        assert_eq!(deserialized.header.timestamp, 1700000000000);
        assert!(
            deserialized.commands.is_empty(),
            "Empty block must have no commands"
        );
    }

    // ---------------------------------------------------------------
    // Integration test: streaming WAL recovery with large unreplicated
    // tail does not OOM and correctly replicates all records to Postgres.
    //
    // Regression test for CALL-654 / PR #850.
    // ---------------------------------------------------------------

    /// Set up a Postgres test database and return the diesel pool.
    ///
    /// Checks `TEST_DATABASE_URL` first to reuse a pre-started Postgres service
    /// (CI). Falls back to a testcontainer for local development.
    async fn setup_test_pool() -> (
        hypercall_db_diesel::DbPool,
        Option<testcontainers::ContainerAsync<testcontainers_modules::postgres::Postgres>>,
        crate::test_contracts::TestnetContractEnvGuard,
    ) {
        use crate::test_contracts::test_database_url_with_database;
        use testcontainers::runners::AsyncRunner;
        use testcontainers::ImageExt;
        use testcontainers_modules::postgres::Postgres;

        let env_guard = crate::test_contracts::TestnetContractEnvGuard::apply();

        let (database_url, container) = if let Ok(url) = std::env::var("TEST_DATABASE_URL") {
            let db_name = format!("test_wal_recovery_{}", uuid::Uuid::new_v4().simple());
            let admin_pool = sqlx::postgres::PgPoolOptions::new()
                .max_connections(1)
                .connect(&url)
                .await
                .expect("connect to TEST_DATABASE_URL");
            sqlx::query(&format!(r#"CREATE DATABASE "{}""#, db_name))
                .execute(&admin_pool)
                .await
                .expect("create isolated test database");
            admin_pool.close().await;
            let db_url = test_database_url_with_database(&url, &db_name);
            (db_url, None)
        } else {
            let container = Postgres::default()
                .with_db_name("test_wal_recovery")
                .with_user("test_user")
                .with_password("test_password")
                .with_tag("16-alpine")
                .start()
                .await
                .expect("start postgres container");

            let port = container
                .get_host_port_ipv4(5432)
                .await
                .expect("get postgres port");

            let url = format!(
                "postgresql://test_user:test_password@127.0.0.1:{}/test_wal_recovery",
                port
            );

            // Wait for Postgres to be ready.
            tokio::time::sleep(tokio::time::Duration::from_secs(2)).await;

            (url, Some(container))
        };

        let _handler = crate::db_handler::DieselEventHandler::new(&database_url)
            .expect("create diesel handler and run migrations");

        let auth = hypercall_db_diesel::DbAuthConfig::password(&database_url);
        let pool = hypercall_db_diesel::build_db_pool(&auth, 5, 30_000, 10_000)
            .expect("build diesel pool");

        (pool, container, env_guard)
    }

    #[tokio::test]
    #[serial_test::serial(testnet_env)]
    async fn recover_wal_sync_large_unreplicated_tail() {
        // Regression test for CALL-654: streaming WAL recovery must process
        // a large unreplicated tail in bounded-memory batches without OOM,
        // replicate every record to PostgreSQL, and advance the checkpoint
        // to the end of the WAL.

        let (pool, _container, _env_guard) = setup_test_pool().await;

        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        // Write 1000 WAL entries to simulate a large unreplicated tail.
        let num_entries: usize = 1000;
        let mut wal = WalState::open_at(wal_path.clone()).expect("open wal");

        let entries: Vec<JournalEntry> = (0..num_entries)
            .map(|i| JournalEntry {
                request_uuid: hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4()),
                received_ts_ms: 1000 + i as u64,
                command_data: vec![(i % 256) as u8; 64],
                response_data: Some(vec![(i % 256) as u8; 32]),
                order_id: Some(i as i64),
                pre_digest: EngineStateDigest::default(),
                post_digest: EngineStateDigest::default(),
                duration_ms: 1,
                events: vec![],
                outbox_appends: vec![],
                fill_side_effects: vec![],
                cash_withdrawal_side_effect: None,
                balance_updates: Vec::new(),
                created_at: Instant::now(),
                commit_ack: None,
                command_type_enum: Some(
                    hypercall_db_diesel::engine_enums::CommandType::CreateOrder,
                ),
                #[cfg(feature = "rsm-state")]
                command_identity_hash: [0u8; 32],
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
            })
            .collect();

        let request_uuids: Vec<hypercall_db_diesel::engine_enums::DbUuid> =
            entries.iter().map(|e| e.request_uuid).collect();

        let offsets = wal.append_batch(&entries).expect("append WAL entries");
        assert_eq!(offsets.len(), num_entries);

        let wal_end_offset = offsets.last().unwrap().end_offset;
        drop(wal);

        // Checkpoint at 0 => entire WAL is unreplicated.
        write_checkpoint(
            &checkpoint_path,
            WalCheckpointMetadata {
                wal_offset: 0,
                last_command_id: 0,
                last_l2_seq: 0,
            },
        )
        .expect("write initial checkpoint");

        // Use a small batch size (10) to force many streaming iterations.
        let small_batch_size: usize = 10;

        // Run streaming WAL recovery.
        let truncated = EngineJournalBatcher::recover_wal_sync(
            &pool,
            &wal_path,
            &checkpoint_path,
            false, // persist_digests
            small_batch_size,
            #[cfg(feature = "rsm-state")]
            hypercall_db::ValidatorRsmEnvironment::Development,
            #[cfg(feature = "rsm-state")]
            None,
        );

        // No torn bytes, so no truncation should occur.
        assert!(
            truncated.is_none(),
            "WAL should not be truncated when all records are complete"
        );

        // Verify checkpoint advanced to end of WAL.
        let checkpoint = read_checkpoint(&checkpoint_path).expect("read checkpoint");
        assert_eq!(
            checkpoint.wal_offset, wal_end_offset,
            "Checkpoint must advance to end of WAL after full recovery"
        );
        assert!(
            checkpoint.last_command_id > 0,
            "Checkpoint must record a positive last_command_id"
        );

        // Verify all 1000 records were inserted into engine_commands.
        let mut conn = pool.get().expect("get db connection");
        let inserted_count: i64 =
            diesel::dsl::sql::<diesel::sql_types::BigInt>("SELECT COUNT(*) FROM engine_commands")
                .get_result(&mut conn)
                .expect("count engine_commands");

        assert_eq!(
            inserted_count, num_entries as i64,
            "All {} WAL entries must be replicated to engine_commands",
            num_entries
        );

        // Verify request_uuids match by spot-checking first, middle, and last.
        for &idx in &[0, num_entries / 2, num_entries - 1] {
            let uuid_val = request_uuids[idx].0;
            let exists: bool = diesel::dsl::sql::<diesel::sql_types::Bool>(&format!(
                "SELECT EXISTS(SELECT 1 FROM engine_commands WHERE request_uuid = '{}')",
                uuid_val
            ))
            .get_result(&mut conn)
            .expect("check request_uuid exists");

            assert!(
                exists,
                "request_uuid for entry {} must exist in engine_commands",
                idx
            );
        }
    }

    #[tokio::test]
    #[serial_test::serial(testnet_env)]
    async fn recover_wal_sync_checkpoint_advances_per_batch() {
        // Verify that the checkpoint advances after each batch, not just at
        // the end. This ensures crash-recovery mid-replay resumes correctly.

        let (pool, _container, _env_guard) = setup_test_pool().await;

        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);

        // Write 50 entries, use batch_size=10 so we get 5 batches.
        let num_entries: usize = 50;
        let mut wal = WalState::open_at(wal_path.clone()).expect("open wal");

        let entries: Vec<JournalEntry> = (0..num_entries)
            .map(|i| JournalEntry {
                request_uuid: hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4()),
                received_ts_ms: 2000 + i as u64,
                command_data: vec![(i % 256) as u8; 16],
                response_data: None,
                order_id: Some(i as i64),
                pre_digest: EngineStateDigest::default(),
                post_digest: EngineStateDigest::default(),
                duration_ms: 1,
                events: vec![],
                outbox_appends: vec![],
                fill_side_effects: vec![],
                cash_withdrawal_side_effect: None,
                balance_updates: Vec::new(),
                created_at: Instant::now(),
                commit_ack: None,
                command_type_enum: Some(
                    hypercall_db_diesel::engine_enums::CommandType::CreateOrder,
                ),
                #[cfg(feature = "rsm-state")]
                command_identity_hash: [0u8; 32],
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
            })
            .collect();

        let offsets = wal.append_batch(&entries).expect("append WAL entries");
        let wal_end_offset = offsets.last().unwrap().end_offset;
        drop(wal);

        // No checkpoint file => defaults to offset 0.
        EngineJournalBatcher::recover_wal_sync(
            &pool,
            &wal_path,
            &checkpoint_path,
            false,
            10,
            #[cfg(feature = "rsm-state")]
            hypercall_db::ValidatorRsmEnvironment::Development,
            #[cfg(feature = "rsm-state")]
            None,
        );

        // After recovery, checkpoint must be at the very end.
        let checkpoint = read_checkpoint(&checkpoint_path).expect("read checkpoint");
        assert_eq!(
            checkpoint.wal_offset, wal_end_offset,
            "Checkpoint must be at end of WAL after complete recovery"
        );

        // Now simulate a partial replay: reset checkpoint to midpoint,
        // then recover again. Idempotent inserts mean no duplicates.
        let midpoint_offset = offsets[24].end_offset;
        write_checkpoint(
            &checkpoint_path,
            WalCheckpointMetadata {
                wal_offset: midpoint_offset,
                last_command_id: checkpoint.last_command_id,
                last_l2_seq: checkpoint.last_l2_seq,
            },
        )
        .expect("write midpoint checkpoint");

        EngineJournalBatcher::recover_wal_sync(
            &pool,
            &wal_path,
            &checkpoint_path,
            false,
            10,
            #[cfg(feature = "rsm-state")]
            hypercall_db::ValidatorRsmEnvironment::Development,
            #[cfg(feature = "rsm-state")]
            None,
        );

        // Checkpoint should again be at end.
        let checkpoint2 =
            read_checkpoint(&checkpoint_path).expect("read checkpoint after re-recovery");
        assert_eq!(checkpoint2.wal_offset, wal_end_offset);

        // Count should still be 50 (idempotent via ON CONFLICT DO NOTHING).
        let mut conn = pool.get().expect("get db connection");
        let count: i64 =
            diesel::dsl::sql::<diesel::sql_types::BigInt>("SELECT COUNT(*) FROM engine_commands")
                .get_result(&mut conn)
                .expect("count engine_commands");

        assert_eq!(
            count, num_entries as i64,
            "Re-recovery must not create duplicate rows"
        );
    }

    #[tokio::test]
    #[serial_test::serial(testnet_env)]
    async fn recover_wal_sync_duplicate_request_id_persists_digest_once() {
        let (pool, _container, _env_guard) = setup_test_pool().await;

        let dir = tempfile::tempdir().expect("create tempdir");
        let wal_path = dir.path().join("journal.wal");
        let checkpoint_path = checkpoint_path_for_wal(&wal_path);
        let duplicate_request_uuid =
            hypercall_db_diesel::engine_enums::DbUuid(uuid::Uuid::new_v4());
        let mut wal = WalState::open_at(wal_path.clone()).expect("open wal");

        let entries: Vec<JournalEntry> = (0..2)
            .map(|i| JournalEntry {
                request_uuid: duplicate_request_uuid,
                received_ts_ms: 3000 + i as u64,
                command_data: vec![i as u8; 16],
                response_data: None,
                order_id: None,
                pre_digest: EngineStateDigest::default(),
                post_digest: EngineStateDigest::default(),
                duration_ms: 1,
                events: vec![EventPayload {
                    event_topic: "test.recovery".to_string(),
                    event_key: Some(format!("duplicate-event-{i}")),
                    event_data: vec![1, i as u8],
                    l2_sequence: Some(i as i64),
                    event_type_enum: hypercall_db_diesel::engine_enums::EventType::OrderbookUpdated,
                }],
                outbox_appends: vec![],
                balance_updates: vec![],
                fill_side_effects: vec![],
                cash_withdrawal_side_effect: None,
                created_at: Instant::now(),
                commit_ack: None,
                command_type_enum: Some(
                    hypercall_db_diesel::engine_enums::CommandType::DepositUpdate,
                ),
                #[cfg(feature = "rsm-state")]
                command_identity_hash: [0u8; 32],
                #[cfg(feature = "rsm-state")]
                rsm_state_digest: Some(crate::rsm::engine_snapshot::EngineStateDigest::empty()),
            })
            .collect();

        let offsets = wal.append_batch(&entries).expect("append WAL entries");
        let wal_end_offset = offsets.last().unwrap().end_offset;
        drop(wal);

        EngineJournalBatcher::recover_wal_sync(
            &pool,
            &wal_path,
            &checkpoint_path,
            true,
            10,
            #[cfg(feature = "rsm-state")]
            hypercall_db::ValidatorRsmEnvironment::Development,
            #[cfg(feature = "rsm-state")]
            None,
        );

        let checkpoint = read_checkpoint(&checkpoint_path).expect("read checkpoint");
        assert_eq!(
            checkpoint.wal_offset, wal_end_offset,
            "duplicate request UUIDs must not block checkpoint advancement"
        );

        let mut conn = pool.get().expect("get db connection");
        let command_count: i64 =
            diesel::dsl::sql::<diesel::sql_types::BigInt>("SELECT COUNT(*) FROM engine_commands")
                .get_result(&mut conn)
                .expect("count engine_commands");
        assert_eq!(
            command_count, 1,
            "duplicate request UUIDs materialize one engine command"
        );

        let event_count: i64 =
            diesel::dsl::sql::<diesel::sql_types::BigInt>("SELECT COUNT(*) FROM engine_events")
                .get_result(&mut conn)
                .expect("count engine_events");
        assert_eq!(
            event_count, 1,
            "duplicate request UUIDs materialize one command event"
        );

        let digest_count: i64 = diesel::dsl::sql::<diesel::sql_types::BigInt>(
            "SELECT COUNT(*) FROM engine_command_digests",
        )
        .get_result(&mut conn)
        .expect("count engine_command_digests");
        assert_eq!(
            digest_count, 1,
            "duplicate request UUIDs materialize one command digest"
        );
    }

    #[cfg(feature = "rsm-state")]
    #[tokio::test]
    #[serial_test::serial(testnet_env)]
    async fn rsm_wal_recovery_rebuilds_local_commitment_store_before_db_roots() {
        let (pool, _container, _env_guard) = setup_test_pool().await;
        let env = hypercall_db::ValidatorRsmEnvironment::Development;
        let dir = tempfile::tempdir().expect("create tempdir");
        let config = ValidatorRsmCommitmentStoreConfig {
            root_dir: dir.path().join("state"),
            prune_window: 100,
            pruning_enabled: false,
        };

        let mut runtime = ValidatorRsmStateCommitmentRuntime::open_from_db(
            config.clone(),
            Arc::new(pool.clone()),
            env,
        )
        .expect("open empty commitment runtime");

        let mut entry = make_test_rsm_entry(1);
        entry.command_identity_hash = [7u8; 32];
        let entries = [entry];
        let mut block = make_test_rsm_block_records(0, [0u8; 32], 0, &entries);

        apply_recovered_rsm_commitment_blocks(&mut block, &mut runtime)
            .expect("replay should apply local commitment roots before DB persistence");
        EngineJournalBatcher::insert_batch(
            &pool,
            &entries,
            false,
            Some(
                EngineJournalBatcher::rsm_block_persistence_batch(RsmBlockPersistenceInput {
                    environment: env,
                    records: &block,
                    mode: RsmBlockPersistenceMode::Replay,
                })
                .expect("prepare RSM replay persistence"),
            ),
        )
        .expect("persist replayed roots");

        drop(runtime);
        ValidatorRsmStateCommitmentRuntime::open_from_db(config, Arc::new(pool.clone()), env)
            .expect("strict DB/local root verification should pass after replay");
    }

    #[cfg(feature = "rsm-state")]
    #[tokio::test]
    #[serial_test::serial(testnet_env)]
    async fn rsm_replay_does_not_rewind_current_pointer() {
        let (pool, _container, _env_guard) = setup_test_pool().await;
        let env = hypercall_db::ValidatorRsmEnvironment::Development;

        let mut entry1 = make_test_rsm_entry(1);
        entry1.command_identity_hash = [1u8; 32];
        let mut entry2 = make_test_rsm_entry(2);
        entry2.command_identity_hash = [2u8; 32];

        let entries = [entry1, entry2];

        let block1 = make_test_rsm_block_records(1, [0u8; 32], 0, &entries[0..1]);
        EngineJournalBatcher::insert_batch(
            &pool,
            &entries[0..1],
            false,
            Some(
                EngineJournalBatcher::rsm_block_persistence_batch(RsmBlockPersistenceInput {
                    environment: env,
                    records: &block1,
                    mode: RsmBlockPersistenceMode::Live,
                })
                .expect("prepare RSM live block 1 persistence"),
            ),
        )
        .expect("persist live block 1");
        let block1_hash = block1[0]
            .block_header
            .as_ref()
            .expect("block 1 header")
            .block_hash();

        let block2 = make_test_rsm_block_records(2, block1_hash, 1, &entries[1..2]);
        EngineJournalBatcher::insert_batch(
            &pool,
            &entries[1..2],
            false,
            Some(
                EngineJournalBatcher::rsm_block_persistence_batch(RsmBlockPersistenceInput {
                    environment: env,
                    records: &block2,
                    mode: RsmBlockPersistenceMode::Live,
                })
                .expect("prepare RSM live block 2 persistence"),
            ),
        )
        .expect("persist live block 2");

        let handler =
            hypercall_db_diesel::DatabaseHandler::with_pool_no_migrations(Arc::new(pool.clone()));
        assert_eq!(
            handler
                .get_validator_rsm_current_state_sync(env)
                .expect("read current")
                .expect("current exists")
                .current_version,
            2
        );

        EngineJournalBatcher::insert_batch(
            &pool,
            &entries[0..1],
            false,
            Some(
                EngineJournalBatcher::rsm_block_persistence_batch(RsmBlockPersistenceInput {
                    environment: env,
                    records: &block1,
                    mode: RsmBlockPersistenceMode::Replay,
                })
                .expect("prepare RSM replay block persistence"),
            ),
        )
        .expect("replay old block without rewinding current");

        assert_eq!(
            handler
                .get_validator_rsm_current_state_sync(env)
                .expect("read current after replay")
                .expect("current exists")
                .current_version,
            2
        );
    }
}