feat: implement full replication subsystem

[mickvandijke]

Summary

• Implements the complete replication design from docs/REPLICATION_DESIGN.md across 14 new source files (8,001 lines in src/replication/)
• Fresh replication with PoP validation, neighbor sync with round-robin cycles, batched quorum verification, storage audit protocol, LMDB-backed PaidForList, responsibility pruning with hysteresis, topology churn handling, bootstrap sync with drain gate
• All 16 design sections implemented, zero stubs/TODOs remaining

New modules

Module	Lines	Design Section
mod.rs	1,711	ReplicationEngine + background tasks + message handlers
protocol.rs	932	Wire messages (postcard serialized)
quorum.rs	932	Batched verification (Section 9)
audit.rs	785	Storage audit challenge-response (Section 15)
paid_list.rs	635	LMDB-backed PaidForList (Invariant 15)
types.rs	607	FSM states, queue entries, domain types
scheduling.rs	566	Pipeline queues (Section 12)
neighbor_sync.rs	503	Round-robin cycle management (Section 6.2)
config.rs	470	All tunable parameters (Section 4)
admission.rs	317	Per-key hint admission (Section 7)
bootstrap.rs	244	Bootstrap sync + drain gate (Section 16)
pruning.rs	162	Post-cycle pruning with hysteresis (Section 11)
fresh.rs	137	Fresh replication with PoP (Section 6.1)

Modified existing files

• node.rs — Integrates ReplicationEngine into RunningNode lifecycle
• storage/handler.rs — Exposes storage() and payment_verifier_arc() accessors
• storage/lmdb.rs — Adds all_keys() and get_raw() for replication
• error.rs — Adds Replication error variant
• lib.rs — Adds module + re-exports

Test plan

• 444 unit tests pass (cargo test --lib)
• Zero clippy warnings (strict mode)
• Release build succeeds
• E2E tests for Section 18 test matrix (in progress)

🤖 Generated with Claude Code

[Copilot]

Pull request overview

Implements a full replication subsystem (per docs/REPLICATION_DESIGN.md) and integrates it into node lifecycle + e2e test harness, adding scheduling, neighbor-sync, quorum verification, audits, pruning, and an LMDB-backed paid list.

Changes:

• Add src/replication/ with wire protocol, engine orchestration, neighbor sync, verification queues, pruning/audit/bootstrapping, and paid-list persistence.
• Integrate ReplicationEngine into RunningNode startup/shutdown and extend storage APIs to support replication/audit.
• Update CI workflows to run unit tests and e2e tests separately.

Reviewed changes

Copilot reviewed 27 out of 27 changed files in this pull request and generated 10 comments.

Show a summary per file

File	Description
src/replication/mod.rs	Replication engine orchestration, background tasks, and message handling integration points.
src/replication/protocol.rs	Defines replication wire messages (postcard) and size limits.
src/replication/config.rs	Centralizes replication tuning parameters and validation helpers.
src/replication/types.rs	Replication FSM + domain types used across the subsystem.
src/replication/scheduling.rs	Pipeline queues (pending verify, fetch queue, in-flight) and eviction/dedup helpers.
src/replication/neighbor_sync.rs	Round-robin neighbor sync ordering, cooldown, and request/response helpers.
src/replication/fresh.rs	Fresh replication fanout + PaidNotify emission.
src/replication/bootstrap.rs	Bootstrap gate/drain tracking helpers for replication startup.
src/replication/pruning.rs	Post-cycle pruning of out-of-range records and paid-list entries with hysteresis.
src/replication/paid_list.rs	LMDB-backed persistent PaidForList plus hysteresis timestamp tracking.
src/replication/audit.rs	Storage-audit challenge/response and digest verification flow.
src/replication/quorum.rs	Batched verification/quorum evaluation and evidence aggregation.
src/node.rs	Wires ReplicationEngine into the node lifecycle and DHT event subscription timing.
src/lib.rs	Exposes replication module and re-exports ReplicationEngine / ReplicationConfig.
src/error.rs	Adds Error::Replication variant.
src/storage/handler.rs	Adds accessors for storage and payment verifier needed by replication.
src/storage/lmdb.rs	Adds all_keys() and get_raw() to support hint construction and audits.
src/storage/mod.rs	Updates protocol ID documentation string.
src/ant_protocol/chunk.rs	Changes CHUNK_PROTOCOL_ID value (wire routing identifier).
tests/e2e/testnet.rs	Starts/stops replication engine in e2e nodes; bumps max message size to accommodate replication traffic.
tests/e2e/data_types/chunk.rs	Adjusts restart simulation to fully shutdown replication before reopening LMDB.
tests/e2e/mod.rs	Adds replication e2e module.
docs/REPLICATION_DESIGN.md	Adds the replication design/specification document.
.github/workflows/ci.yml	Splits unit vs e2e test execution.
.github/workflows/release.yml	Splits unit vs e2e test execution in release workflow.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

In the neighbor-sync round, peers that respond with bootstrapping=true are still treated as a successful sync: record_successful_sync is called and sync_history is updated. This can incorrectly create RepairOpportunity for a peer that explicitly said it is not ready, and it also puts the peer under cooldown even though no bidirectional sync occurred. Consider handling resp.bootstrapping before recording success (track bootstrap_claims/abuse, remove the peer from the current snapshot, and do not update last_sync_times / sync_history).

[Copilot]

On sync failure, handle_sync_failure returns a replacement peer to fill the vacated slot, but the result is ignored. This means rounds can run with fewer than neighbor_sync_peer_count peers even when additional peers remain in the snapshot, which conflicts with the intended “fill the vacated slot” behavior. Consider scheduling the returned replacement (and applying the same cooldown/eligibility checks as batch selection).

[Copilot]

enqueue_fetch does not check pending_verify, so a key can be present in both PendingVerify and FetchQueue if a caller enqueues before removing from pending. This contradicts the module-level guarantee of strict dedup across all pipeline stages. Consider either (a) making enqueue_fetch also return early when pending_verify contains the key, or (b) providing a single transition API that atomically removes from pending and enqueues for fetch.

[Copilot]

PruneResult.records_marked_out_of_range / records_cleared are documented as counting timestamps “newly set/cleared”, but the current logic increments these counters unconditionally for every in-range / out-of-range key, even if the timestamp was already in the desired state. Consider only incrementing when set_record_out_of_range actually inserts a new timestamp (and when clear_record_out_of_range actually removes one), so the reported counts match their docs.

Suggested change

	paid_list.clear_record_out_of_range(key);
	result.records_cleared += 1;
	} else {
	paid_list.set_record_out_of_range(key);
	result.records_marked_out_of_range += 1;
	// Only clear and count if there was previously an out-of-range timestamp.
	if paid_list.record_out_of_range_since(key).is_some() {
	paid_list.clear_record_out_of_range(key);
	result.records_cleared += 1;
	}
	} else {
	// Only set and count if this record was not already marked out of range.
	let was_out_of_range = paid_list.record_out_of_range_since(key).is_some();
	if !was_out_of_range {
	paid_list.set_record_out_of_range(key);
	result.records_marked_out_of_range += 1;
	}

[Copilot]

PruneResult.paid_entries_marked / paid_entries_cleared are documented as counting timestamps “newly set/cleared”, but the current logic increments these counters for every key regardless of whether a timestamp was newly inserted/removed. Consider aligning the counters with the actual state changes (e.g., have PaidList::{set,clear}_paid_out_of_range return whether a mutation occurred).

Suggested change

	if in_paid_group {
	paid_list.clear_paid_out_of_range(key);
	result.paid_entries_cleared += 1;
	} else {
	paid_list.set_paid_out_of_range(key);
	result.paid_entries_marked += 1;
	// Inspect prior out-of-range state so that counters reflect actual transitions.
	let was_out_of_range = paid_list.paid_out_of_range_since(key).is_some();
	if in_paid_group {
	// Entry is (or should be) in range; clear timestamp only if it was set.
	if was_out_of_range {
	paid_list.clear_paid_out_of_range(key);
	result.paid_entries_cleared += 1;
	}
	} else {
	// Entry is out of our paid group; ensure it's marked out of range.
	paid_list.set_paid_out_of_range(key);
	if !was_out_of_range {
	result.paid_entries_marked += 1;
	}

[Copilot]

VerificationState is described as matching the Section 8 state-transition diagram, but the design doc includes a FilterRejected state that does not exist in this enum. This makes the code/doc mapping ambiguous for readers and tests that reference the spec. Consider either adding a FilterRejected variant (and using it where admission fails) or updating the spec/comments to reflect the implemented FSM.

[Copilot]

MAX_REPLICATION_MESSAGE_SIZE is defined both here and in replication/config.rs, with a comment that they “match”. Duplicating the constant risks drift if either value changes later. Consider having a single source of truth (e.g., re-export the config constant from this module, or vice versa) and referencing it directly.

Suggested change

	/// Maximum replication wire message size (10 MB).
	///
	/// Accommodates hint batches and record payloads with envelope overhead.
	/// Matches `config::MAX_REPLICATION_MESSAGE_SIZE`.
	const MAX_MESSAGE_SIZE_MIB: usize = 10;
	/// Maximum replication wire message size in bytes.
	pub const MAX_REPLICATION_MESSAGE_SIZE: usize = MAX_MESSAGE_SIZE_MIB * 1024 * 1024;
	/// Maximum replication wire message size in bytes.
	///
	/// Accommodates hint batches and record payloads with envelope overhead.
	/// Re-exported from `config::MAX_REPLICATION_MESSAGE_SIZE` to ensure a single
	/// source of truth for the limit.
	pub use super::config::MAX_REPLICATION_MESSAGE_SIZE;

[Copilot]

handle_sync_failure is documented as resuming the Rule-2 scan to fill a vacated slot, but it simply returns the next peer at the current cursor without applying cooldown filtering/removal. This can select a peer that should have been skipped due to cooldown, and it also diverges from the spec comment. Consider either reusing select_sync_batch to obtain replacements (so the same rules apply) or passing the cooldown parameters into handle_sync_failure and applying the same filtering logic.

[Copilot]

Changing CHUNK_PROTOCOL_ID changes the protocol string used for routing chunk messages. This is a wire-compatibility breaking change for any existing nodes/clients still using the previous ID. Consider calling this out explicitly in the PR description/release notes and/or providing a migration/compatibility strategy (e.g., supporting both IDs for a deprecation window or bumping protocol/versioning accordingly).

[Copilot]

all_keys() performs a full LMDB scan synchronously on the calling thread. Since it’s used from async contexts (neighbor sync, pruning, audit), scanning a large store here can block the Tokio runtime. Consider making this an async API that uses spawn_blocking (similar to get/delete) or otherwise ensure calls are confined to blocking threads / capped to avoid long runtime stalls.

Suggested change

	pub fn all_keys(&self) -> Result<Vec<XorName>> {
	let rtxn = self
	.env
	.read_txn()
	.map_err(|e| Error::Storage(format!("Failed to create read txn: {e}")))?;
	let mut keys = Vec::new();
	let iter = self
	.db
	.iter(&rtxn)
	.map_err(|e| Error::Storage(format!("Failed to iterate database: {e}")))?;
	for result in iter {
	let (key_bytes, _) =
	result.map_err(|e| Error::Storage(format!("Failed to read entry: {e}")))?;
	if key_bytes.len() == 32 {
	let mut key = [0u8; 32];
	key.copy_from_slice(key_bytes);
	keys.push(key);
	}
	}
	Ok(keys)
	pub async fn all_keys(&self) -> Result<Vec<XorName>> {
	let env = self.env.clone();
	let db = self.db;
	let handle = spawn_blocking(move || {
	let rtxn = env
	.read_txn()
	.map_err(|e| Error::Storage(format!("Failed to create read txn: {e}")))?;
	let mut keys = Vec::new();
	let iter = db
	.iter(&rtxn)
	.map_err(|e| Error::Storage(format!("Failed to iterate database: {e}")))?;
	for result in iter {
	let (key_bytes, _) =
	result.map_err(|e| Error::Storage(format!("Failed to read entry: {e}")))?;
	if key_bytes.len() == 32 {
	let mut key = [0u8; 32];
	key.copy_from_slice(key_bytes);
	keys.push(key);
	}
	}
	Ok::<Vec<XorName>, Error>(keys)
	});
	handle
	.await
	.map_err(|e| Error::Storage(format!("Failed to join all_keys task: {e}")))?