Struct EngineJournalBatcher 

pub struct EngineJournalBatcher {
    pool: DbPool,
    receiver: Receiver<JournalMessage>,
    config: JournalBatcherConfig,
    shutdown: Receiver<()>,
    wal_writer_tx: SyncSender<WalWriterMessage>,
    wal_writer_handle: Option<JoinHandle<()>>,
    replication_tx: Sender<ReplicationMessage>,
    replication_rx: Option<Receiver<ReplicationMessage>>,
    rsm_environment: ValidatorRsmEnvironment,
}

The batcher that runs in the background and appends to local WAL first, then replicates to PostgreSQL asynchronously.

Fields

pool: DbPool

receiver: Receiver<JournalMessage>

config: JournalBatcherConfig

shutdown: Receiver<()>

wal_writer_tx: SyncSender<WalWriterMessage>

Sender half of the bounded channel to the dedicated WAL writer thread.

wal_writer_handle: Option<JoinHandle<()>>

Join handle for the WAL writer OS thread (taken during shutdown).

replication_tx: Sender<ReplicationMessage>

replication_rx: Option<Receiver<ReplicationMessage>>

rsm_environment: ValidatorRsmEnvironment

Environment partition for persisted RSM roots and block headers.

Implementations

impl EngineJournalBatcher

pub(super) fn rsm_block_persistence_batch( input: RsmBlockPersistenceInput<'_>, ) -> Result<EngineJournalRsmBlockBatch>

impl EngineJournalBatcher

pub fn new( pool: DbPool, config: JournalBatcherConfig, shutdown: Receiver<()>, cold_signer: Option<PrivateKeySigner>, ) -> (Self, JournalBatchSender)

Create a new batcher and return the sender for pushing entries.

fn load_rsm_block_cursor( pool: &DbPool, rsm_environment: ValidatorRsmEnvironment, ) -> (u64, [u8; 32], u64)

fn recover_wal_sync( pool: &DbPool, wal_path: &Path, checkpoint_path: &Path, persist_digests: bool, max_batch_size: usize, rsm_environment: ValidatorRsmEnvironment, rsm_commitments: Option<&mut ValidatorRsmStateCommitmentRuntime>, ) -> Option<u64>

Run streaming WAL recovery. Returns Some(truncated_len) if the WAL was truncated to remove torn bytes, so the caller can fix up WalState.

pub async fn run(self)

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.

async fn flush_batch( &mut self, batch: &mut Vec<JournalEntry>, completion: Option<Sender<()>>, )

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 send_flush_barrier(&self, completion: Sender<()>)

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 await_replication_flush(&self)

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 replication_loop( pool: DbPool, persist_digests: bool, max_batch_size: usize, flush_interval: Duration, checkpoint_path: PathBuf, wal_path: PathBuf, receiver: Receiver<ReplicationMessage>, rsm_environment: ValidatorRsmEnvironment, )

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.

fn read_checkpoint_or_panic(path: &Path) -> WalCheckpointMetadata

fn drain_replication_channel( receiver: &mut Receiver<ReplicationMessage>, pending: &mut Vec<WalRecordWithOffset>, )

fn punch_wal_hole(wal_path: &Path, from: u64, to: u64)

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 advance_checkpoint( checkpoint_path: &Path, last_offset: u64, progress: ReplicationProgress, ) -> Result<(), String>

fn emit_wal_checkpoint_metrics(wal_path: &Path, checkpoint_offset: u64)

Emit Prometheus gauges for WAL checkpoint offset and unreplicated tail.

async fn replicate_pending( pool: &DbPool, persist_digests: bool, max_batch_size: usize, checkpoint_path: &Path, pending: &mut Vec<WalRecordWithOffset>, rsm_environment: ValidatorRsmEnvironment, ) -> Result<(), String>

async fn replicate_chunk( pool: &DbPool, persist_digests: bool, records: &[WalRecordWithOffset], rsm_environment: ValidatorRsmEnvironment, ) -> Result<ReplicationProgress, String>

fn insert_batch( pool: &DbPool, entries: &[JournalEntry], persist_digests: bool, rsm_blocks: Option<EngineJournalRsmBlockBatch>, ) -> Result<(usize, Vec<(DbUuid, i64)>)>

Insert a batch of entries into PostgreSQL (blocking). Returns (inserted_count, [(request_uuid, command_id)]) for downstream event publish accounting.

fn lookup_command_ids_by_request_ids( pool: &DbPool, request_ids: &[DbUuid], ) -> Result<Vec<(DbUuid, i64)>>

pub fn start(self) -> JoinHandle<()>

Start the batcher as a background task