feat(replication): primary → replica data sync stream

[kacy]

summary

implements the replication stream between primary and replica nodes (phase 6, pr 2). when a node runs CLUSTER REPLICATE <id>, it connects to the primary's replication port and loads a full-state snapshot, then receives all subsequent mutations as a live stream.

replication.rs — new module with two types:

• ReplicationServer: accepts TCP connections from replicas. sends a per-shard binary snapshot (using the existing snapshot format in-memory) followed by an unbounded stream of AofRecord frames. on broadcast channel lag, sends MSG_RESYNC and closes; the replica reconnects and resync's
• ReplicationClient: connects to the primary, loads the snapshot into the local engine via key-based routing, replays incremental records, and reconnects with exponential backoff (500ms → 30s) on any disconnect

wire protocol (little-endian framing):

replica → primary: [version: 1B][num_shards: 2B]
primary → replica: [version: 1B][num_shards: 2B][id_len: 1B][id: N bytes][status: 1B]
per shard:         [MSG_SHARD_SYNC][shard_id: 2B][len: 4B][snapshot bytes]
                   [MSG_SHARD_OFFSET][shard_id: 2B][offset: 8B]
incremental:       [MSG_RECORD][shard_id: 2B][offset: 8B][len: 4B][AofRecord bytes]
on lag:            [MSG_RESYNC]  → replica reconnects

cluster.rs — adds set_engine, start_replication_server, start_replication_client, and replication_info. cluster_replicate now connects to the primary after topology update. replication port = data_port + gossip_offset + 2.

server.rs / main.rs — broadcast channel (65536 capacity) is created when cluster mode is enabled, injected into EngineConfig, and the engine is set on the coordinator after startup so the replication server can use it.

connection.rs — INFO replication section with role, connected_replicas (primary) or master_host/port/link_status (replica).

what was tested

• cargo test --workspace: all unit tests pass. cluster integration tests show the same pre-existing failures as main (12, down from 14 — two tests now pass that didn't before)
• cargo clippy --workspace -- -D warnings: clean
• cargo build --workspace: clean

design considerations

the broadcast channel is the bridge between shard mutations and the replication stream. each shard publishes ReplicationEvent { shard_id, offset, record } after every successful write. the server subscribes a new receiver per replica connection, so each replica gets an independent stream starting from the current position. snapshots are serialized in-memory using io::Cursor<Vec<u8>> — avoids disk I/O for full-sync, which matters on fast reconnects.

replica routing uses the same FNV-1a key-hash as the primary (via engine.route(key, request)), so records always land on the correct shard without a separate shard-ID lookup.