Crate replication_engine 

Replication Engine

A mesh replication agent for synchronizing data between sync-engine nodes.

Architecture

The replication engine sits between the local sync-engine and remote peers, managing bidirectional data flow via Redis Streams:

┌───────────────────────────────────────────────────────────────────────────┐
│                          replication-engine                               │
│                                                                           │
│  ┌─────────────┐    ┌──────────────┐    ┌──────────────────────────────┐  │
│  │ PeerManager │───►│ StreamTailer │───►│ Dedup + Apply to sync-engine │  │
│  │ (per peer)  │    │ (XREAD)      │    │ (hash comparison)            │  │
│  └─────────────┘    └──────────────┘    └──────────────────────────────┘  │
│         │                                            │                    │
│         ▼                                            ▼                    │
│  ┌─────────────┐                          ┌─────────────────────────┐     │
│  │ CursorStore │                          │ MerkleRepair (cold path)│     │
│  │ (SQLite)    │                          │ (periodic anti-entropy) │     │
│  └─────────────┘                          └─────────────────────────┘     │
└───────────────────────────────────────────────────────────────────────────┘

Two-Path Replication

1. Hot Path (CDC Streams): Real-time tailing of peer cdc streams
2. Cold Path (Merkle Repair): Periodic anti-entropy using Merkle tree comparison

Usage

use replication_engine::{ReplicationEngine, ReplicationEngineConfig};
use tokio::sync::watch;

#[tokio::main]
async fn main() {
    let config = ReplicationEngineConfig::default();
    let (_tx, rx) = watch::channel(config.clone());
     
    let mut engine = ReplicationEngine::new(config, rx);
    engine.start().await.expect("Failed to start");
     
    // Engine runs until shutdown signal
    engine.shutdown().await;
}

Re-exports

pub use circuit_breaker::CircuitBreaker;

pub use circuit_breaker::CircuitConfig;

pub use circuit_breaker::CircuitError;

pub use circuit_breaker::SyncEngineCircuit;

pub use config::ReplicationEngineConfig;

pub use config::ReplicationEngineSettings;

pub use config::PeerConfig;

pub use config::HotPathConfig;

pub use config::ColdPathConfig;

pub use config::CursorConfig;

pub use config::SloConfig;

pub use config::PeerHealthConfig;

pub use coordinator::ReplicationEngine;

pub use coordinator::EngineState;

pub use coordinator::HealthCheck;

pub use coordinator::PeerHealth;

pub use error::ReplicationError;

pub use error::Result;

pub use cursor::CursorStore;

pub use peer::PeerManager;

pub use peer::PeerCircuitState;

pub use stream::StreamTailer;

pub use stream::CdcEvent;

pub use sync_engine::SyncEngineRef;

pub use sync_engine::NoOpSyncEngine;

pub use sync_engine::SyncError;

Modules

batch

Batch processor for CDC events.

circuit_breaker

Circuit breaker pattern for sync-engine protection.

config

Configuration for the replication engine.

coordinator

Replication engine coordinator.

cursor

Cursor persistence for stream positions.

error

Error types for the replication engine.

metrics

Metrics for observability.

peer

Peer connection management.

resilience

Resilience utilities: retry logic, backoff, rate limiting, bulkheads.

stream

Redis Stream consumer for CDC events.

sync_engine

Sync engine integration traits.