Struct ReplicationEngine 

pub struct ReplicationEngine<S: SyncEngineRef = NoOpSyncEngine> { /* private fields */ }

The main replication engine.

Manages bidirectional data sync between this node and its mesh peers.

Sync Engine Integration

The replication engine is passed a reference to the local sync-engine by the daemon. We use this to:

• Write replicated data from peers (submit)
• Check for duplicates before applying (is_current)
• Query Merkle tree for cold path repair

We never write to any CDC stream — that’s sync-engine’s responsibility. We only read from peer CDC streams and write to local sync-engine.

Implementations

impl ReplicationEngine<NoOpSyncEngine>

pub fn new( config: ReplicationEngineConfig, config_rx: Receiver<ReplicationEngineConfig>, ) -> Self

Create a new replication engine with no-op sync engine (for testing/standalone).

The engine starts in Created state. Call start() to connect to peers and begin replication.

impl<S: SyncEngineRef> ReplicationEngine<S>

pub fn with_sync_engine( config: ReplicationEngineConfig, config_rx: Receiver<ReplicationEngineConfig>, sync_engine: Arc<S>, ) -> Self

Create a new replication engine with a sync-engine reference.

This is the primary constructor used by the daemon.

Arguments

• config - Replication configuration
• config_rx - Watch channel for config updates
• sync_engine - Reference to local sync-engine (for writes and dedup)

pub fn sync_engine(&self) -> &Arc<S>

Get a reference to the sync engine.

pub fn circuit(&self) -> &Arc<SyncEngineCircuit>

Get a reference to the circuit breaker for sync-engine protection.

pub fn state(&self) -> EngineState

Get current engine state.

pub fn state_receiver(&self) -> Receiver<EngineState>

Get a receiver to watch state changes.

pub fn is_running(&self) -> bool

Check if engine is running.

pub async fn health_check(&self) -> HealthCheck

Get comprehensive health status for monitoring endpoints.

Returns a HealthCheck struct containing:

• Engine state and readiness
• Sync-engine backpressure status
• Per-peer connectivity, circuit breaker state, and lag

Performance: This method performs no network I/O. All data is collected from cached internal state (atomics, mutexes, watch channels).

Example

let health = engine.health_check().await;
 
// For /ready endpoint
if health.ready {
    HttpResponse::Ok()
} else {
    HttpResponse::ServiceUnavailable()
}

// For /health endpoint (full diagnostics)
HttpResponse::Ok().json(serde_json::json!({
    "healthy": health.healthy,
    "state": health.state.to_string(),
    "peers_connected": health.peers_connected,
    "peers_total": health.peers_total,
    "sync_engine_accepting_writes": health.sync_engine_accepting_writes,
}))

pub async fn start(&mut self) -> Result<()>

Start the replication engine.

1. Opens cursor store (SQLite)
2. Connects to all enabled peers
3. Spawns hot path tailers for each peer
4. Spawns cold path repair task (if enabled)

pub async fn shutdown(&mut self)

Shutdown the replication engine gracefully.

Shutdown sequence:

1. Signal all hot/cold path tasks to stop
2. Wait for tasks to flush pending batches (with timeout)
3. Shutdown peer connections
4. Checkpoint and close cursor store

pub fn peer_manager(&self) -> &Arc<PeerManager>

Get the peer manager (for metrics/diagnostics).

pub fn node_id(&self) -> &str

Get the node ID.