feat: implement full replication subsystem

.github/workflows/ci.yml

@@ -47,8 +47,10 @@ jobs:
         uses: foundry-rs/foundry-toolchain@v1
         with:
           version: nightly
-      - name: Run tests
-        run: cargo test
+      - name: Run unit tests
+        run: cargo test --lib --features test-utils
+      - name: Run e2e tests
+        run: cargo test --test e2e --features test-utils -- --test-threads=1
 
   doc:
     name: Documentation

.github/workflows/release.yml

@@ -100,8 +100,10 @@ jobs:
       - name: Run clippy
         run: cargo clippy --all-targets --all-features -- -D warnings
 
-      - name: Run tests
-        run: cargo test
+      - name: Run unit tests
+        run: cargo test --lib --features test-utils
+      - name: Run e2e tests
+        run: cargo test --test e2e --features test-utils -- --test-threads=1
 
   build:
     name: Build ${{ matrix.target }}

src/ant_protocol/chunk.rs

@@ -9,7 +9,7 @@
 use serde::{Deserialize, Serialize};
 
 /// Protocol identifier for chunk operations.
-pub const CHUNK_PROTOCOL_ID: &str = "autonomi/ant/chunk/v1";
+pub const CHUNK_PROTOCOL_ID: &str = "autonomi.ant.chunk.v1";
 
 /// Current protocol version.
 pub const PROTOCOL_VERSION: u16 = 1;
@@ -519,7 +519,7 @@ mod tests {
 
     #[test]
     fn test_constants() {
-        assert_eq!(CHUNK_PROTOCOL_ID, "autonomi/ant/chunk/v1");
+        assert_eq!(CHUNK_PROTOCOL_ID, "autonomi.ant.chunk.v1");
         assert_eq!(PROTOCOL_VERSION, 1);
         assert_eq!(MAX_CHUNK_SIZE, 4 * 1024 * 1024);
         assert_eq!(DATA_TYPE_CHUNK, 0);

src/error.rs

@@ -52,6 +52,10 @@ pub enum Error {
     #[error("invalid chunk: {0}")]
     InvalidChunk(String),
 
+    /// Replication error.
+    #[error("replication error: {0}")]
+    Replication(String),
+
     /// Node is shutting down.
     #[error("node is shutting down")]
     ShuttingDown,

src/lib.rs

@@ -47,6 +47,7 @@ pub mod error;
 pub mod event;
 pub mod node;
 pub mod payment;
+pub mod replication;
 pub mod storage;
 pub mod upgrade;
 
@@ -65,6 +66,7 @@ pub use error::{Error, Result};
 pub use event::{NodeEvent, NodeEventsChannel};
 pub use node::{NodeBuilder, RunningNode};
 pub use payment::{PaymentStatus, PaymentVerifier, PaymentVerifierConfig};
+pub use replication::{config::ReplicationConfig, ReplicationEngine};
 pub use storage::{AntProtocol, LmdbStorage, LmdbStorageConfig};
 
 /// Re-exports from `saorsa-core` so downstream crates (e.g. `ant-client`)

src/node.rs

@@ -10,6 +10,8 @@ use crate::event::{create_event_channel, NodeEvent, NodeEventsChannel, NodeEvent
 use crate::payment::metrics::QuotingMetricsTracker;
 use crate::payment::wallet::parse_rewards_address;
 use crate::payment::{EvmVerifierConfig, PaymentVerifier, PaymentVerifierConfig, QuoteGenerator};
+use crate::replication::config::ReplicationConfig;
+use crate::replication::ReplicationEngine;
 use crate::storage::{AntProtocol, LmdbStorage, LmdbStorageConfig};
 use crate::upgrade::{
     upgrade_cache_dir, AutoApplyUpgrader, BinaryCache, ReleaseCache, UpgradeMonitor, UpgradeResult,
@@ -133,15 +135,43 @@ impl NodeBuilder {
             None
         };
 
+        let p2p_arc = Arc::new(p2p_node);
+
+        // Initialize replication engine (if storage is enabled)
+        let replication_engine = if let Some(ref protocol) = ant_protocol {
+            let repl_config = ReplicationConfig::default();
+            let storage_arc = protocol.storage();
+            let payment_verifier_arc = protocol.payment_verifier_arc();
+            match ReplicationEngine::new(
+                repl_config,
+                Arc::clone(&p2p_arc),
+                storage_arc,
+                payment_verifier_arc,
+                &self.config.root_dir,
+                shutdown.clone(),
+            )
+            .await
+            {
+                Ok(engine) => Some(engine),
+                Err(e) => {
+                    warn!("Failed to initialize replication engine: {e}");
+                    None
+                }
+            }
+        } else {
+            None
+        };
+
         let node = RunningNode {
             config: self.config,
-            p2p_node: Arc::new(p2p_node),
+            p2p_node: p2p_arc,
             shutdown,
             events_tx,
             events_rx: Some(events_rx),
             upgrade_monitor,
             bootstrap_manager,
             ant_protocol,
+            replication_engine,
             protocol_task: None,
             upgrade_exit_code: Arc::new(AtomicI32::new(-1)),
         };
@@ -431,6 +461,8 @@ pub struct RunningNode {
     bootstrap_manager: Option<BootstrapManager>,
     /// ANT protocol handler for chunk storage.
     ant_protocol: Option<Arc<AntProtocol>>,
+    /// Replication engine (manages neighbor sync, verification, audits).
+    replication_engine: Option<ReplicationEngine>,
     /// Protocol message routing background task.
     protocol_task: Option<JoinHandle<()>>,
     /// Exit code requested by a successful upgrade (-1 = no upgrade exit pending).
@@ -466,6 +498,14 @@ impl RunningNode {
     pub async fn run(&mut self) -> Result<()> {
         info!("Node runtime loop starting");
 
+        // Subscribe to DHT events BEFORE starting the P2P node so the
+        // bootstrap-sync task does not miss the BootstrapComplete event
+        // emitted during P2PNode::start().
+        let dht_events_for_bootstrap = self
+            .replication_engine
+            .as_ref()
+            .map(|_| self.p2p_node.dht_manager().subscribe_events());
+
         // Start the P2P node
         self.p2p_node
             .start()
@@ -493,6 +533,16 @@ impl RunningNode {
         // Start protocol message routing (P2P → AntProtocol → P2P response)
         self.start_protocol_routing();
 
+        // Start replication engine background tasks
+        if let Some(ref mut engine) = self.replication_engine {
+            // Safety: dht_events_for_bootstrap is Some when replication_engine
+            // is Some (both arms use the same condition).
+            if let Some(dht_events) = dht_events_for_bootstrap {
+                engine.start(dht_events);
+            }
+            info!("Replication engine started");
+        }
+
         // Start upgrade monitor if enabled
         if let Some(monitor) = self.upgrade_monitor.take() {
             let events_tx = self.events_tx.clone();
@@ -652,6 +702,12 @@ impl RunningNode {
             );
         }
 
+        // Shutdown replication engine before P2P so background tasks don't
+        // use a dead P2P layer, and Arc<LmdbStorage> references are released.
+        if let Some(ref mut engine) = self.replication_engine {
+            engine.shutdown().await;
+        }
+
         // Stop protocol routing task
         if let Some(handle) = self.protocol_task.take() {
             handle.abort();