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;
@@ -30,6 +30,9 @@ pub const DATA_TYPE_CHUNK: u32 = 0;
 /// Content-addressed identifier (32 bytes).
 pub type XorName = [u8; 32];
 
+/// Byte length of an [`XorName`].
+pub const XORNAME_LEN: usize = std::mem::size_of::<XorName>();
+
 /// Enum of all chunk protocol message types.
 ///
 /// Uses a single-byte discriminant for efficient wire encoding.
@@ -519,7 +522,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/ant_protocol/mod.rs

@@ -62,5 +62,5 @@ pub use chunk::{
     ChunkPutResponse, ChunkQuoteRequest, ChunkQuoteResponse, MerkleCandidateQuoteRequest,
     MerkleCandidateQuoteResponse, ProtocolError, XorName, CHUNK_PROTOCOL_ID, DATA_TYPE_CHUNK,
     MAX_CHUNK_SIZE, MAX_WIRE_MESSAGE_SIZE, PROOF_TAG_MERKLE, PROOF_TAG_SINGLE_NODE,
-    PROTOCOL_VERSION,
+    PROTOCOL_VERSION, XORNAME_LEN,
 };

src/config.rs

@@ -133,9 +133,11 @@ pub struct NodeConfig {
     /// Maximum application-layer message size in bytes.
     ///
     /// Tunes the QUIC stream receive window and per-stream read buffer.
-    /// Default: [`MAX_WIRE_MESSAGE_SIZE`](crate::ant_protocol::MAX_WIRE_MESSAGE_SIZE)
-    /// (5 MiB — sufficient for 4 MiB data chunks plus serialization
-    /// envelope overhead).
+    /// Default: the larger of
+    /// [`MAX_WIRE_MESSAGE_SIZE`](crate::ant_protocol::MAX_WIRE_MESSAGE_SIZE) (5 MiB)
+    /// and [`MAX_REPLICATION_MESSAGE_SIZE`](crate::replication::config::MAX_REPLICATION_MESSAGE_SIZE)
+    /// (10 MiB), so both chunk and replication traffic fit within the transport
+    /// ceiling.
     #[serde(default = "default_max_message_size")]
     pub max_message_size: usize,
 
@@ -350,7 +352,11 @@ pub fn default_nodes_dir() -> PathBuf {
 }
 
 fn default_max_message_size() -> usize {
-    crate::ant_protocol::MAX_WIRE_MESSAGE_SIZE
+    // Use the larger of the chunk protocol and replication protocol ceilings
+    // so that replication hint batches (up to 10 MiB) are not silently dropped
+    // by the transport layer.
+    crate::replication::config::MAX_REPLICATION_MESSAGE_SIZE
+        .max(crate::ant_protocol::MAX_WIRE_MESSAGE_SIZE)
 }
 
 fn default_log_level() -> String {

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,
@@ -123,25 +125,57 @@ impl NodeBuilder {
             None
         };
 
-        // Initialize ANT protocol handler for chunk storage
-        let ant_protocol = if self.config.storage.enabled {
-            Some(Arc::new(
-                Self::build_ant_protocol(&self.config, &identity).await?,
-            ))
+        // Initialize ANT protocol handler for chunk storage and
+        // wire the fresh-write channel so PUTs trigger replication.
+        let (ant_protocol, fresh_write_rx) = if self.config.storage.enabled {
+            let (fresh_write_tx, fresh_write_rx) = tokio::sync::mpsc::unbounded_channel();
+            let mut protocol = Self::build_ant_protocol(&self.config, &identity).await?;
+            protocol.set_fresh_write_sender(fresh_write_tx);
+            (Some(Arc::new(protocol)), Some(fresh_write_rx))
         } else {
             info!("Chunk storage disabled");
-            None
+            (None, None)
         };
 
+        let p2p_arc = Arc::new(p2p_node);
+
+        // Initialize replication engine (if storage is enabled)
+        let replication_engine =
+            if let (Some(ref protocol), Some(fresh_rx)) = (&ant_protocol, fresh_write_rx) {
+                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,
+                    fresh_rx,
+                    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 +465,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 +502,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 +537,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 +706,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();