Module replication 

Replication — synchronous input-stream streaming from primary to replica.

The primary’s JournalStage encodes each InputSlot it just durably journaled into a wire-ready InputBatch frame in the replication ring (separately from the journal-codec bytes it writes to disk). The ReplicationSender thread forwards those frames as-is over TCP/DPDK — no decode + re-encode on the hot path. The replica decodes the frames straight back into InputSlots, publishes them to its local input disruptor with the primary’s pre-assigned sequences and timestamps, and the replica’s JournalStage re-encodes them through its own writer for byte-exact-on-replay durability.

Wire Protocol

Length-prefixed frames, little-endian, over a dedicated TCP connection (or DPDK pipe). The full InputBatch payload layout lives in melin_transport_core::replication::protocol / melin_transport_core::replication_wire.

Auth (before handshake)

• Challenge (Primary → Replica): [len:u32][0x03][nonce:[u8;32]]
• ChallengeResponse (Replica → Primary): [len:u32][0x04][signature:[u8;64]][pubkey:[u8;32]]
• AuthOk (Primary → Replica): [len:u32][0x05]
• AuthFailed (Primary → Replica): [len:u32][0x06]

Replica → Primary

• Handshake: [len:u32][0x01][last_sequence:u64][chain_hash:[u8;32]]
• Ack: [len:u32][0x02][acked_sequence:u64][in_memory_sequence:u64]

Primary → Replica

• StreamStart: [len:u32][0x10][start_sequence:u64][genesis_len:u32][genesis_bytes...]
• NeedSnapshot: [len:u32][0x11]
• HashMismatch: [len:u32][0x12]
• SnapshotBegin: [len:u32][0x13][snapshot_len:u64][snap_sequence:u64][snap_chain_hash:[u8;32]]
• SnapshotChunk: [len:u32][0x14][data...]
• SnapshotEnd: [len:u32][0x15][crc32c:u32]
• InputBatch: [len:u32][0x21][count:u16][slot...] — see transport-core::replication_wire for the per-slot layout
• Heartbeat: [len:u32][0x30][sequence:u64]

v1 Limitations

• No handshake chain hash validation (HashMismatch never sent)
• Dual replication (up to 2 replicas in parallel)

See docs/replication.md for the full design document and limitation details.

Structs

Ack

Ack message sent by the replica.

Handshake

Handshake message sent by the replica on connection.

PendingAck

Pending ack waiting for journal durability confirmation.

PendingAckQueue

Fixed-capacity circular buffer of pending acks. Decouples TCP receives from journal fsync by allowing up to cap batches to be submitted to the journal stage before any ack is sent. Acks are flushed in FIFO order as the journal cursor advances.

ReplicationMetrics

Per-slot replication metrics exposed via the health endpoint.

Sender

Owned state for the replication sender thread.

Enums

PrimaryMessage

Messages from primary to replica.

ReplicaMessage

Messages from replica to primary.

Functions

run_receiver

run_sender

Run the replication sender. Listens for replica connections, streams journal data batches, processes acks, and updates the replication cursor.

try_flush_dual_track

Build the next replica → primary Ack to fire under the dual-track model. Returns Some(Ack) when one of the two cursor tracks has advanced beyond the last value sent on the wire.

update_dual_replication_cursor

Update both replication cursors after an ack or disconnect.

wait_for_journal_cursor

Wait until the journal cursor crosses target. busy_spin=true keeps the wait on the CPU (production default — ack RTT is on the primary’s response-gate critical path, where a yield_now scheduler tick adds ~1ms to client p99). busy_spin=false yields after a short spin so a stalled cursor doesn’t peg a core under --yield-idle (tests / CI / shared boxes).

Type Aliases

ReceiverResult

Outcome of run_receiver: None = clean shutdown, Some = promotion.