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][segment_start_sequence:u64][anchor_hash:[u8;32]][epoch:u64][ack_policy:u8][genesis_known:u8][genesis_entries:u64] — the segment header identity a fresh replica creates its journal with (lineage origin for full catch-up, the seeded segment’s identity after a snapshot transfer), and the lineage’s genesis length, which a fresh replica records and one with a journal checks against its own
• NeedSnapshot: [len:u32][0x11]
• HashMismatch: [len:u32][0x12] — divergent replica journal; the replica archives its lineage, then the snapshot flow follows
• 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]
• Rotate: [len:u32][0x16][boundary_seq:u64][tail_hash:[u8;32]] — primary-driven rotation, verified + adopted by the replica
• ChainCheck: [len:u32][0x17][sequence:u64][chain_hash:[u8;32]] — periodic live-stream chain validation
• SegmentSeedBegin: [len:u32][0x18][seed_len:u64] — raw byte prefix of the primary’s segment containing the snapshot boundary; body rides SnapshotChunk frames, ends with SnapshotEnd
• InputBatch: [len:u32][0x21][count:u16][slot...] — see transport-core::replication_wire for the per-slot layout
• Heartbeat: [len:u32][0x30][sequence:u64]

Limitations

• 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.

ReplicaControlPlane

The handles a replica’s receive loop shares with the rest of the node — the admin endpoint and the control-plane raft driver. Bundled (rather than five positional parameters) because they always travel together through the kernel-TCP and DPDK receiver signatures, and a transposition between same-typed flags is exactly the bug a bundle prevents.

ReplicaCursors

Single owner of the primary’s per-replica progress cursors.

ReplicationListener

Owned state for the replication sender thread. A TCP listener guaranteed non-blocking: the only constructor sets the flag, so the guarantee is carried by the type. run_sender’s accept loop polls the shutdown flag between accepts — handed a blocking listener it would park inside accept() and only observe shutdown when the next replica happened to dial in. This newtype makes that state unrepresentable instead of documenting it away.

ReplicationMetrics

Per-slot replication metrics exposed via the health endpoint.

Sender

SentHighWater

Highest primary sequence streamed to one replica connection.

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.

wait_for_journal_cursor

Wait until the journal cursor crosses target, or abort flips (returns false — the cursor’s owner has died and the wait would never complete). Under BusySpin the wait stays on the CPU: ack RTT is on the primary’s response-gate critical path, where a scheduler tick adds ~1ms to client p99. Under SpinThenYield a stalled cursor doesn’t peg a core (tests / CI / shared boxes).

Type Aliases

ReceiverResult

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