diff --git a/crates/node/src/node.rs b/crates/node/src/node.rs index 0f1038e9e..0c9186e95 100644 --- a/crates/node/src/node.rs +++ b/crates/node/src/node.rs @@ -5,7 +5,7 @@ use crate::{ ZoneEngine, - replication::{broadcast_persisted_blocks, import_leader_blocks}, + replication::{broadcast_persisted_blocks, run_block_sync}, rpc::{ZoneRpc, ZoneRpcApi, rpc_connection_config, start_private_rpc}, }; use alloy_primitives::Address; @@ -557,25 +557,17 @@ where events, } = handle.into_parts(); - match role { - Role::Leader => { - task_executor.spawn_critical_task( - "zone-p2p-block-broadcast", - broadcast_persisted_blocks(provider, commands), - ); - // Leaders do not receive block messages. Dropping this receiver is harmless: the - // runtime only emits BlockReceived on followers. - drop(events); - } - Role::Follower => { - // Keep the command sender alive so the runtime's command loop remains available - // for later ACK/backfill commands even though followers send nothing in this PR. - task_executor.spawn_critical_task( - "zone-p2p-block-import", - import_leader_blocks(provider, engine, events, commands), - ); - } + if role == Role::Leader { + // Only a leader can build + broadcast blocks + task_executor.spawn_critical_task( + "zone-p2p-block-broadcast", + broadcast_persisted_blocks(provider.clone(), commands.clone()), + ); } + task_executor.spawn_critical_task( + "zone-p2p-block-sync", + run_block_sync(provider, engine, events, commands), + ); task_executor.spawn_critical_with_graceful_shutdown_signal( "zone-p2p", diff --git a/crates/node/src/replication.rs b/crates/node/src/replication.rs index da3ecd453..036c7cafb 100644 --- a/crates/node/src/replication.rs +++ b/crates/node/src/replication.rs @@ -6,10 +6,11 @@ use alloy_rlp::Decodable as _; use alloy_rpc_types_engine::ForkchoiceState; use futures::{StreamExt as _, stream::BoxStream}; use reth_chain_state::PersistedBlockSubscriptions; -use reth_node_api::PayloadTypes as _; -use reth_node_builder::ConsensusEngineHandle; +use reth_node_api::{ConsensusEngineHandle, PayloadTypes as _}; use reth_primitives_traits::SealedBlock; -use reth_storage_api::{BlockNumReader, BlockReader, HeaderProvider}; +use reth_provider::HeaderProvider; +use reth_storage_api::{BlockNumReader, BlockReader}; +use std::{collections::BTreeMap, time::Duration}; use tempo_primitives::{Block, TempoHeader}; use tokio::sync::mpsc; use tracing::{debug, info}; @@ -156,29 +157,367 @@ where Ok(()) } -/// Decode, fully execute, and canonicalize blocks received by a follower. -pub(crate) async fn import_leader_blocks

( +const BACKFILL_RETRY_INTERVAL: Duration = Duration::from_secs(1); +const BLOCK_INACTIVITY_TIMEOUT: Duration = Duration::from_secs(30); +const MAX_PENDING_BLOCKS: usize = 128; +const BACKFILL_PAGE_SIZE: u64 = 64; +const BACKFILL_SERVE_QUEUE_CAPACITY: usize = 8; + +struct BackfillRequest { + peer: zone_p2p::P2pPeerId, + request_id: u64, + start: u64, +} + +struct BackfillServerTask(tokio::task::JoinHandle<()>); + +impl Drop for BackfillServerTask { + fn drop(&mut self) { + self.0.abort(); + } +} + +/// Keep track of the backfill exactly. We'll buffer any live blocks received +/// during backfill. +struct BackfillProgress { + target_tip: Option, + received_completion: bool, + needed: bool, +} + +impl BackfillProgress { + const fn new() -> Self { + Self { + target_tip: None, + received_completion: false, + needed: true, + } + } + + fn observe_block(&mut self, number: u64, best: u64) { + self.target_tip = Some(self.target_tip.map_or(number, |tip| tip.max(number))); + if number > best.saturating_add(1) { + self.needed = true; + } + } + + fn refresh_after_import(&mut self, best: u64, first_pending: Option) { + self.needed = !self.received_completion + || self.target_tip.is_some_and(|tip| best < tip) + || first_pending.is_some_and(|number| number > best.saturating_add(1)); + } + + fn complete(&mut self, tip: u64, best: u64, first_pending: Option) { + self.received_completion = true; + self.target_tip = Some(self.target_tip.map_or(tip, |target| target.max(tip))); + self.needed = best < self.target_tip.unwrap_or(tip) + || first_pending.is_some_and(|number| number > best.saturating_add(1)); + } + + fn request(&self, best: u64) -> Option { + self.needed.then(|| P2pCommand::RequestBackfill { + start: best.saturating_add(1), + }) + } + + fn probe_after_inactivity(&mut self, best: u64) -> P2pCommand { + self.needed = true; + P2pCommand::RequestBackfill { + start: best.saturating_add(1), + } + } +} + +fn buffer_pending_block( + pending: &mut BTreeMap>, + number: u64, + block: Vec, +) -> Option { + if pending.contains_key(&number) { + return None; + } + if pending.len() < MAX_PENDING_BLOCKS { + pending.insert(number, block); + return None; + } + + let Some((&farthest, _)) = pending.last_key_value() else { + pending.insert(number, block); + return None; + }; + if number < farthest { + pending.pop_last(); + pending.insert(number, block); + Some(farthest) + } else { + Some(number) + } +} + +fn encoded_block_number(encoded: &[u8]) -> eyre::Result { + let mut input = encoded; + let block = Block::decode(&mut input) + .map_err(|err| eyre::eyre!("invalid RLP-encoded zone block: {err}"))?; + eyre::ensure!( + input.is_empty(), + "encoded zone block has {} trailing bytes", + input.len() + ); + Ok(block.header.number()) +} + +fn serve_backfill_page

( + provider: &P, + commands: &mpsc::Sender, + peer: zone_p2p::P2pPeerId, + request_id: u64, + start: u64, +) -> eyre::Result<()> +where + P: BlockNumReader + BlockReader, +{ + let tip = provider.best_block_number()?; + let end = tip.min(start.saturating_add(BACKFILL_PAGE_SIZE.saturating_sub(1))); + for number in start..=end { + let block = provider.block_by_number(number)?.ok_or_else(|| { + eyre::eyre!("canonical block {number} is missing while serving backfill") + })?; + commands + .blocking_send(P2pCommand::SendBackfillBlock { + peer: peer.clone(), + request_id, + block: alloy_rlp::encode(block), + }) + .map_err(|_| eyre::eyre!("P2P command channel closed"))?; + } + commands + .blocking_send(P2pCommand::CompleteBackfill { + peer, + request_id, + tip, + }) + .map_err(|_| eyre::eyre!("P2P command channel closed"))?; + Ok(()) +} + +async fn serve_backfill_requests

( + provider: P, + commands: mpsc::Sender, + mut requests: mpsc::Receiver, +) where + P: BlockNumReader + BlockReader + Clone + Send + Sync + 'static, +{ + // One worker deliberately serializes page construction and sending, bounding serving + // concurrency independently of the block import loop. + while let Some(BackfillRequest { + peer, + request_id, + start, + }) = requests.recv().await + { + let page_provider = provider.clone(); + let page_commands = commands.clone(); + let page = tokio::task::spawn_blocking(move || { + serve_backfill_page(&page_provider, &page_commands, peer, request_id, start) + }) + .await; + let result = match page { + Ok(result) => result, + Err(err) => Err(eyre::eyre!("backfill page worker failed: {err}")), + }; + if let Err(err) = result { + tracing::error!(target: "zone::p2p", %err, start, "Failed serving block backfill"); + } + } +} + +/// Serve catch-up requests and import live/backfilled blocks in canonical order. +pub(crate) async fn run_block_sync

( provider: P, engine: ConsensusEngineHandle, mut events: mpsc::Receiver, - _commands: mpsc::Sender, + commands: mpsc::Sender, ) where - P: BlockNumReader + HeaderProvider

+ Clone + Send + Sync + 'static, + P: BlockNumReader + + BlockReader + + HeaderProvider
+ + Clone + + Send + + Sync + + 'static, { - while let Some(event) = events.recv().await { - let P2pEvent::BlockReceived { block, .. } = event else { - continue; - }; - if let Err(err) = import_leader_block(&provider, &engine, &block).await { - tracing::error!(target: "zone::p2p", %err, "Rejected leader block"); + // Keep track of all live blocks if we end up needing a backfill, so we can immediately catchup + // This is capped to `MAX_PENDING_BLOCKS`. + let mut pending = BTreeMap::>::new(); + let mut backfill = BackfillProgress::new(); + let (backfill_requests, backfill_request_rx) = mpsc::channel(BACKFILL_SERVE_QUEUE_CAPACITY); + + // Serve backfill requests in a separate task to avoid competing the live blocks + let mut backfill_server = BackfillServerTask(tokio::spawn(serve_backfill_requests( + provider.clone(), + commands.clone(), + backfill_request_rx, + ))); + + // Always probe on startup to see if we're behind + let mut retry = tokio::time::interval(BACKFILL_RETRY_INTERVAL); + retry.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + // If we don't get blocks after 30s, probe to get the next block + let inactivity = tokio::time::sleep(BLOCK_INACTIVITY_TIMEOUT); + tokio::pin!(inactivity); + + loop { + tokio::select! { + event = events.recv() => { + let Some(event) = event else { + debug!(target: "zone::p2p", "P2P event channel closed"); + return; + }; + match event { + P2pEvent::Started { .. } => {} + P2pEvent::BackfillRequested { peer, request_id, start } => { + if let Err(err) = backfill_requests.try_send(BackfillRequest { peer, request_id, start }) { + tracing::warn!(target: "zone::p2p", %err, start, queue_capacity = BACKFILL_SERVE_QUEUE_CAPACITY, "Dropped block backfill request because the serving queue is unavailable"); + } + } + P2pEvent::BlockReceived { block, .. } + | P2pEvent::BackfillBlockReceived { block, .. } => { + match encoded_block_number(&block) { + Ok(number) => { + inactivity + .as_mut() + .reset(tokio::time::Instant::now() + BLOCK_INACTIVITY_TIMEOUT); + let best = match provider.best_block_number() { + Ok(best) => best, + Err(err) => { + tracing::error!(target: "zone::p2p", %err, "Failed reading local head"); + continue; + } + }; + if number <= best { + if let Err(err) = import_peer_block(&provider, &engine, &block).await { + tracing::error!(target: "zone::p2p", %err, "Rejected duplicate or conflicting peer block"); + } + continue; + } + backfill.observe_block(number, best); + if let Some(dropped) = buffer_pending_block(&mut pending, number, block) { + tracing::warn!(target: "zone::p2p", dropped, pending_limit = MAX_PENDING_BLOCKS, "Dropped far-future peer block because the pending block buffer is full"); + } + if number > best.saturating_add(1) { + info!(target: "zone::p2p", local_head = best, received = number, "Detected zone block gap; requesting backfill"); + } + if let Err(err) = drain_pending_blocks(&provider, &engine, &mut pending).await { + tracing::error!(target: "zone::p2p", %err, "Rejected peer block while draining backfill"); + backfill.needed = true; + } else { + let best = match provider.best_block_number() { + Ok(best) => best, + Err(err) => { + tracing::error!(target: "zone::p2p", %err, "Failed reading local head after importing peer blocks"); + continue; + } + }; + backfill.refresh_after_import( + best, + pending.first_key_value().map(|(&number, _)| number), + ); + } + } + Err(err) => tracing::error!(target: "zone::p2p", %err, "Rejected malformed peer block"), + } + } + P2pEvent::BackfillCompleted { peer, tip } => { + let best = match provider.best_block_number() { + Ok(best) => best, + Err(err) => { + tracing::error!(target: "zone::p2p", %err, "Failed reading local head after backfill response"); + continue; + } + }; + backfill.complete( + tip, + best, + pending.first_key_value().map(|(&number, _)| number), + ); + debug!(target: "zone::p2p", %peer, best, tip, backfill_needed = backfill.needed, "Completed block backfill response page"); + } + } + } + _ = retry.tick(), if backfill.needed => { + let best = match provider.best_block_number() { + Ok(best) => best, + Err(err) => { + tracing::error!(target: "zone::p2p", %err, "Failed reading local head for backfill request"); + continue; + } + }; + + // retry the backfill + if let Some(command) = backfill.request(best) + && commands.send(command).await.is_err() + { + debug!(target: "zone::p2p", "P2P command channel closed"); + return; + } + } + _ = &mut inactivity, if !backfill.needed => { + // Reset before reading the provider so a transient provider error cannot leave an + // elapsed sleep continuously ready and spin this loop. + inactivity + .as_mut() + .reset(tokio::time::Instant::now() + BLOCK_INACTIVITY_TIMEOUT); + let best = match provider.best_block_number() { + Ok(best) => best, + Err(err) => { + tracing::error!(target: "zone::p2p", %err, "Failed reading local head for inactivity backfill probe"); + continue; + } + }; + let command = backfill.probe_after_inactivity(best); + info!(target: "zone::p2p", best, "No peer block received recently; probing for backfill"); + if commands.send(command).await.is_err() { + debug!(target: "zone::p2p", "P2P command channel closed"); + return; + } + } + result = &mut backfill_server.0 => { + match result { + Ok(()) => tracing::error!(target: "zone::p2p", "Block backfill server stopped unexpectedly"), + Err(err) => tracing::error!(target: "zone::p2p", %err, "Block backfill server task failed"), + } + return; + } } } - debug!(target: "zone::p2p", "P2P event channel closed"); } -async fn import_leader_block

( +async fn drain_pending_blocks

( + provider: &P, + engine: &reth_node_builder::ConsensusEngineHandle, + pending: &mut BTreeMap>, +) -> eyre::Result<()> +where + P: reth_storage_api::BlockNumReader + + reth_storage_api::HeaderProvider

+ + Clone + + Send + + Sync + + 'static, +{ + loop { + let next = provider.best_block_number()?.saturating_add(1); + let Some(block) = pending.remove(&next) else { + return Ok(()); + }; + import_peer_block(provider, engine, &block).await?; + } +} + +async fn import_peer_block

( provider: &P, - engine: &ConsensusEngineHandle, + engine: &reth_node_builder::ConsensusEngineHandle, encoded: &[u8], ) -> eyre::Result<()> where @@ -203,11 +542,11 @@ where eyre::eyre!("missing local canonical header at height {block_number}") })?; if existing.hash() == hash { - debug!(target: "zone::p2p", block_number, ?hash, "Ignoring duplicate leader block"); + debug!(target: "zone::p2p", block_number, ?hash, "Ignoring duplicate peer block"); return Ok(()); } eyre::bail!( - "leader block conflicts with canonical block at height {block_number}: local={}, received={hash}", + "peer block conflicts with canonical block at height {block_number}: local={}, received={hash}", existing.hash() ); } @@ -215,7 +554,7 @@ where let expected_number = best_block.saturating_add(1); if block_number != expected_number { eyre::bail!( - "leader block gap: local head is {best_block}, received height {block_number}, expected {expected_number}; backfill is not implemented yet" + "peer block gap: local head is {best_block}, received height {block_number}, expected {expected_number}" ); } @@ -225,7 +564,7 @@ where .ok_or_else(|| eyre::eyre!("missing local canonical head at height {best_block}"))?; if block.parent_hash() != parent.hash() { eyre::bail!( - "leader block parent mismatch at height {block_number}: local={}, received={}", + "peer block parent mismatch at height {block_number}: local={}, received={}", parent.hash(), block.parent_hash() ); @@ -235,7 +574,7 @@ where let payload = ZonePayloadTypes::block_to_payload(block, None); let status = engine.new_payload(payload).await?; if !status.is_valid() { - eyre::bail!("execution engine rejected leader block {block_number} ({hash}): {status:?}"); + eyre::bail!("execution engine rejected peer block {block_number} ({hash}): {status:?}"); } // 4. Forkchoice @@ -261,7 +600,8 @@ mod tests { use futures::{StreamExt as _, stream}; use super::{ - EncodedPersistedBlock, PersistedBlockSource, PersistedTip, broadcast_persisted_blocks, + BackfillProgress, EncodedPersistedBlock, MAX_PENDING_BLOCKS, PersistedBlockSource, + PersistedTip, broadcast_persisted_blocks, buffer_pending_block, }; use alloy_primitives::B256; use zone_p2p::P2pCommand; @@ -315,4 +655,85 @@ mod tests { ); assert_eq!(command_rx.recv().await, None); } + + #[test] + fn requests_backfill_again_when_live_block_reveals_gap_after_completion() { + const LOCAL_HEAD: u64 = 10; + + let mut backfill = BackfillProgress::new(); + assert_eq!( + backfill.request(LOCAL_HEAD), + Some(P2pCommand::RequestBackfill { + start: LOCAL_HEAD + 1, + }) + ); + + // The first response catches the follower up to the responder's snapshot tip. + backfill.complete(LOCAL_HEAD, LOCAL_HEAD, None); + assert_eq!(backfill.request(LOCAL_HEAD), None); + + // N+1's live broadcast is missed, then N+2 arrives and remains buffered behind the gap. + let live_block = LOCAL_HEAD + 2; + backfill.observe_block(live_block, LOCAL_HEAD); + backfill.refresh_after_import(LOCAL_HEAD, Some(live_block)); + + // The retry starts at the missing N+1 rather than skipping to the received N+2. + assert_eq!( + backfill.request(LOCAL_HEAD), + Some(P2pCommand::RequestBackfill { + start: LOCAL_HEAD + 1, + }) + ); + } + + #[test] + fn inactivity_probe_restarts_backfill_retries() { + const LOCAL_HEAD: u64 = 10; + + let mut backfill = BackfillProgress::new(); + backfill.complete(LOCAL_HEAD, LOCAL_HEAD, None); + assert_eq!(backfill.request(LOCAL_HEAD), None); + + assert_eq!( + backfill.probe_after_inactivity(LOCAL_HEAD), + P2pCommand::RequestBackfill { + start: LOCAL_HEAD + 1, + } + ); + assert_eq!( + backfill.request(LOCAL_HEAD), + Some(P2pCommand::RequestBackfill { + start: LOCAL_HEAD + 1, + }) + ); + } + + #[test] + fn pending_block_limit_keeps_blocks_closest_to_local_head() { + let mut pending = std::collections::BTreeMap::new(); + for number in 100..100 + MAX_PENDING_BLOCKS as u64 { + assert_eq!( + buffer_pending_block(&mut pending, number, vec![number as u8]), + None + ); + } + assert_eq!(pending.len(), MAX_PENDING_BLOCKS); + + let farthest = 100 + MAX_PENDING_BLOCKS as u64 - 1; + assert_eq!( + buffer_pending_block(&mut pending, 99, vec![99]), + Some(farthest) + ); + assert_eq!(pending.len(), MAX_PENDING_BLOCKS); + assert!(pending.contains_key(&99)); + assert!(!pending.contains_key(&farthest)); + + let farther = farthest + 1; + assert_eq!( + buffer_pending_block(&mut pending, farther, vec![farther as u8]), + Some(farther) + ); + assert_eq!(pending.len(), MAX_PENDING_BLOCKS); + assert!(!pending.contains_key(&farther)); + } } diff --git a/crates/p2p/README.md b/crates/p2p/README.md index 3da3ee4e6..0eec1e984 100644 --- a/crates/p2p/README.md +++ b/crates/p2p/README.md @@ -105,3 +105,16 @@ explicit `--p2p.bypass-ip-check` flag. The flag disables source-IP filtering for all inbound P2P connections, not only the DNS peer. Only use it when a network-level policy restricts the P2P port to the configured peers; Ed25519 manifest membership authentication remains enforced. + +## Block catch-up + +Every node probes for missing blocks when P2P starts and retries while its eligible peers are +offline or a gap remains. Followers request blocks from the statically configured leader. A +recovering leader requests blocks from the configured followers. Both roles can serve bounded +64-block response pages from their persisted canonical chain. + +Backfilled blocks use the same RLP representation and import path as live replicated blocks. A +node buffers out-of-order arrivals, then re-executes and canonicalizes only the next block after +its local head. Parent linkage, execution results, block hash, and forkchoice validation therefore +remain mandatory during catch-up; authenticated transport alone never makes a returned block +canonical. diff --git a/crates/p2p/src/lib.rs b/crates/p2p/src/lib.rs index a26cd1b0c..fb2a999d6 100644 --- a/crates/p2p/src/lib.rs +++ b/crates/p2p/src/lib.rs @@ -8,4 +8,6 @@ mod runtime; pub use manifest::{ManifestAddress, ManifestError, ManifestNode, Role, ZoneManifest}; pub use network::P2pNetworkId; -pub use runtime::{P2pCommand, P2pConfig, P2pEvent, P2pHandle, P2pHandleParts, spawn_p2p}; +pub use runtime::{ + P2pCommand, P2pConfig, P2pEvent, P2pHandle, P2pHandleParts, P2pPeerId, spawn_p2p, +}; diff --git a/crates/p2p/src/network.rs b/crates/p2p/src/network.rs index 3b366adf7..e6f026e3e 100644 --- a/crates/p2p/src/network.rs +++ b/crates/p2p/src/network.rs @@ -11,6 +11,10 @@ use crate::ZoneManifest; /// Leader-to-follower sealed block replication channel. pub(crate) const BLOCK_CHANNEL: u64 = 0; +/// Catch-up range requests sent by a lagging node to eligible peers. +pub(crate) const BACKFILL_REQUEST_CHANNEL: u64 = 1; +/// Ordered block and completion frames returned for a catch-up request. +pub(crate) const BACKFILL_RESPONSE_CHANNEL: u64 = 2; pub(crate) const BLOCK_BACKLOG: usize = 128; // At 30M gas, calldata is bounded below 7.5 MiB; leave headroom for block overhead. @@ -96,6 +100,14 @@ pub(crate) fn block_quota() -> Quota { Quota::per_second(NZU32!(128)) } +pub(crate) fn backfill_request_quota() -> Quota { + Quota::per_second(NZU32!(1)) +} + +pub(crate) fn backfill_response_quota() -> Quota { + Quota::per_second(NZU32!(128)) +} + #[cfg(test)] mod tests { use alloy_primitives::address; diff --git a/crates/p2p/src/runtime.rs b/crates/p2p/src/runtime.rs index 8abb150eb..7ad35af28 100644 --- a/crates/p2p/src/runtime.rs +++ b/crates/p2p/src/runtime.rs @@ -1,25 +1,127 @@ -use std::{net::SocketAddr, path::Path, sync::Arc, time::Duration}; +use std::{ + collections::HashMap, + net::SocketAddr, + path::Path, + sync::Arc, + time::{Duration, Instant}, +}; use commonware_cryptography::ed25519::PublicKey; use commonware_p2p::{ AddressableManager as _, Receiver as _, Recipients, Sender as _, authenticated::lookup, }; use commonware_runtime::{Runner as _, Spawner as _}; -use tokio::sync::{mpsc, oneshot}; +use tokio::sync::{Mutex, mpsc, oneshot}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; use crate::{ P2pNetworkId, Role, ZoneManifest, identity::Ed25519Identity, - network::{self, BLOCK_BACKLOG, BLOCK_CHANNEL, MAX_MESSAGE_SIZE}, + network::{ + self, BACKFILL_REQUEST_CHANNEL, BACKFILL_RESPONSE_CHANNEL, BLOCK_BACKLOG, BLOCK_CHANNEL, + MAX_MESSAGE_SIZE, + }, }; const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5); const BROADCAST_RETRY_INTERVAL: Duration = Duration::from_secs(1); const BROADCAST_RETRY_TIMEOUT: Duration = Duration::from_secs(30); +const BACKFILL_RESPONSE_TIMEOUT: Duration = Duration::from_secs(5 * 60); const COMMAND_BACKLOG: usize = 128; const EVENT_BACKLOG: usize = 128; +const BACKFILL_BLOCK_FRAME: u8 = 0; +const BACKFILL_COMPLETE_FRAME: u8 = 1; + +/// Authenticated Commonware identity used to address one manifest peer. +pub type P2pPeerId = PublicKey; + +type CommonwareSender = lookup::Sender; +type CommonwareReceiver = lookup::Receiver; +type SharedBackfillLifecycle = Arc>; + +#[derive(Debug, Clone, Copy)] +struct OutstandingBackfill { + request_id: u64, + sent_at: Instant, +} + +#[derive(Debug, Default)] +struct BackfillJob { + next_request_id: u64, + outstanding: HashMap, +} + +impl BackfillJob { + fn begin_request( + &mut self, + peers: &[PublicKey], + now: Instant, + ) -> Option<(u64, Vec)> { + let request_peers = peers + .iter() + .filter(|peer| { + self.outstanding.get(*peer).is_none_or(|request| { + now.duration_since(request.sent_at) >= BACKFILL_RESPONSE_TIMEOUT + }) + }) + .cloned() + .collect::>(); + if request_peers.is_empty() { + return None; + } + + let request_id = self.next_request_id; + self.next_request_id = self.next_request_id.wrapping_add(1); + for peer in &request_peers { + self.outstanding.insert( + peer.clone(), + OutstandingBackfill { + request_id, + sent_at: now, + }, + ); + } + Some((request_id, request_peers)) + } + + fn finish_send(&mut self, request_id: u64, sent: &[PublicKey]) { + self.outstanding + .retain(|peer, request| request.request_id != request_id || sent.contains(peer)); + } + + fn cancel_request(&mut self, request_id: u64) { + self.outstanding + .retain(|_, request| request.request_id != request_id); + } + + fn accepts(&self, peer: &PublicKey, request_id: u64, now: Instant) -> bool { + self.outstanding.get(peer).is_some_and(|request| { + request.request_id == request_id + && now.duration_since(request.sent_at) < BACKFILL_RESPONSE_TIMEOUT + }) + } + + fn complete(&mut self, peer: &PublicKey, request_id: u64, now: Instant) -> bool { + if !self.accepts(peer, request_id, now) { + return false; + } + self.outstanding.remove(peer); + true + } +} + +struct P2pSenders { + blocks: CommonwareSender, + backfill_requests: CommonwareSender, + backfill_responses: CommonwareSender, +} + +struct P2pReceivers { + blocks: CommonwareReceiver, + backfill_requests: CommonwareReceiver, + backfill_responses: CommonwareReceiver, +} /// Fully validated configuration for one node's Zone P2P runtime. #[derive(Clone)] @@ -103,6 +205,20 @@ fn validate_ip_check_configuration( pub enum P2pCommand { /// Broadcast one RLP-encoded sealed zone block to all configured followers. BroadcastBlock(Vec), + /// Ask the role-appropriate peers for canonical blocks beginning at `start`. + RequestBackfill { start: u64 }, + /// Return one canonical block to the peer that requested it. + SendBackfillBlock { + peer: PublicKey, + request_id: u64, + block: Vec, + }, + /// Finish one page of a backfill response and advertise the responder's snapshot tip. + CompleteBackfill { + peer: PublicKey, + request_id: u64, + tip: u64, + }, } /// Observable lifecycle and block events emitted by the P2P runtime. @@ -119,6 +235,16 @@ pub enum P2pEvent { leader_ed25519_public_key: PublicKey, block: Vec, }, + /// An authenticated peer requested canonical blocks beginning at `start`. + BackfillRequested { + peer: PublicKey, + request_id: u64, + start: u64, + }, + /// A sealed canonical block was returned by an eligible backfill peer. + BackfillBlockReceived { peer: PublicKey, block: Vec }, + /// The responder sent all blocks available in this response page. + BackfillCompleted { peer: PublicKey, tip: u64 }, } /// Handle used to communicate with, supervise, and stop the dedicated P2P runtime. @@ -243,8 +369,20 @@ fn run( network_id, )?; oracle.track(0, peers).await; - let (sender, receiver) = + let (block_sender, block_receiver) = commonware.register(BLOCK_CHANNEL, network::block_quota(), BLOCK_BACKLOG); + + // The backfill request and responses are on separate channels + let (backfill_request_sender, backfill_request_receiver) = commonware.register( + BACKFILL_REQUEST_CHANNEL, + network::backfill_request_quota(), + BLOCK_BACKLOG, + ); + let (backfill_response_sender, backfill_response_receiver) = commonware.register( + BACKFILL_RESPONSE_CHANNEL, + network::backfill_response_quota(), + BLOCK_BACKLOG, + ); let mut network_task = commonware.start(); if config.bypass_ip_check { @@ -253,6 +391,7 @@ fn run( "P2P source-IP filtering is disabled; relying on network-level access controls and manifest Ed25519 public keys" ); } + info!( target: "zone::p2p", zone_id = config.manifest.zone_id(), @@ -262,6 +401,7 @@ fn run( peers = config.manifest.nodes().len(), "Started P2P networking" ); + let _ = events .send(P2pEvent::Started { role: config.role, @@ -270,20 +410,46 @@ fn run( }) .await; - let followers = config + let followers: Vec = config .manifest .nodes() .iter() .filter(|node| config.manifest.role_of(node.ed25519_public_key()) == Some(Role::Follower)) .map(|node| node.ed25519_public_key().clone()) .collect(); - let command_loop = run_commands(config.role, followers, sender, command_rx); + + let backfill_peers = match config.role { + // A recovering leader can backfill from all followers + Role::Leader => followers.clone(), + + // A recovering follower can backfill from the canonical leader + Role::Follower => vec![config.manifest.leader_ed25519_public_key().clone()], + }; + + let backfill_lifecycle = Arc::new(Mutex::new(BackfillJob::default())); + let command_loop = run_commands( + config.role, + followers, + backfill_peers, + backfill_lifecycle.clone(), + P2pSenders { + blocks: block_sender, + backfill_requests: backfill_request_sender, + backfill_responses: backfill_response_sender, + }, + command_rx, + ); tokio::pin!(command_loop); - let receive_loop = run_block_receiver( + let receive_loop = run_receivers( config.role, config.manifest, - receiver, + P2pReceivers { + blocks: block_receiver, + backfill_requests: backfill_request_receiver, + backfill_responses: backfill_response_receiver, + }, + backfill_lifecycle, events, ); tokio::pin!(receive_loop); @@ -310,85 +476,209 @@ fn run( async fn run_commands( role: Role, followers: Vec, - mut sender: lookup::Sender, + backfill_peers: Vec, + backfill_job: SharedBackfillLifecycle, + mut senders: P2pSenders, mut commands: mpsc::Receiver, ) -> eyre::Result<()> { while let Some(command) = commands.recv().await { - if role != Role::Leader { - warn!(target: "zone::p2p", ?command, "Ignoring outbound block command on follower"); - continue; - } - let P2pCommand::BroadcastBlock(block) = command; - if block.len() > MAX_MESSAGE_SIZE as usize { - error!( - target: "zone::p2p", - block_size_bytes = block.len(), - max_message_size_bytes = MAX_MESSAGE_SIZE, - "Canonical block exceeds the P2P message size limit; block was not broadcast" - ); - continue; - } - let sent = tokio::time::timeout(BROADCAST_RETRY_TIMEOUT, async { - loop { - let sent = sender - .send(Recipients::Some(followers.clone()), block.clone(), true) + match command { + P2pCommand::BroadcastBlock(block) => { + if role != Role::Leader { + warn!(target: "zone::p2p", "Ignoring live block broadcast command on follower"); + continue; + } + + if block.len() > MAX_MESSAGE_SIZE as usize { + error!(target: "zone::p2p", block_size_bytes = block.len(), max_message_size_bytes = MAX_MESSAGE_SIZE, "Canonical block exceeds the P2P message size limit; block was not broadcast"); + continue; + } + + let sent = tokio::time::timeout(BROADCAST_RETRY_TIMEOUT, async { + loop { + let sent = senders.blocks + .send(Recipients::Some(followers.clone()), block.clone(), true) + .await + .map_err(|err| eyre::eyre!("failed broadcasting zone block: {err}"))?; + if !sent.is_empty() || followers.is_empty() { + return Ok::<_, eyre::Report>(sent); + } + debug!(target: "zone::p2p", "No followers are connected; retrying canonical block broadcast"); + tokio::time::sleep(BROADCAST_RETRY_INTERVAL).await; + } + }).await; + let sent = match sent { + Ok(sent) => sent?, + Err(_) => { + warn!(target: "zone::p2p", timeout_secs = BROADCAST_RETRY_TIMEOUT.as_secs(), "No followers connected before block broadcast timed out"); + continue; + } + }; + if sent.len() != followers.len() { + debug!(target: "zone::p2p", connected = sent.len(), configured = followers.len(), "Some followers are not connected; block was not sent to them"); + } + } + + P2pCommand::RequestBackfill { start } => { + let now = Instant::now(); + let request = { + backfill_job + .lock() + .await + .begin_request(&backfill_peers, now) + }; + let Some((request_id, request_peers)) = request else { + debug!(target: "zone::p2p", start, configured = backfill_peers.len(), "Skipping block backfill request because all eligible peers already have outstanding responses"); + continue; + }; + let mut request_frame = Vec::with_capacity(16); + request_frame.extend_from_slice(&request_id.to_be_bytes()); + request_frame.extend_from_slice(&start.to_be_bytes()); + let sent = match senders + .backfill_requests + .send(Recipients::Some(request_peers.clone()), request_frame, true) .await - .map_err(|err| eyre::eyre!("failed broadcasting zone block: {err}"))?; - if !sent.is_empty() || followers.is_empty() { - return Ok::<_, eyre::Report>(sent); + { + Ok(sent) => sent, + Err(err) => { + backfill_job.lock().await.cancel_request(request_id); + return Err(eyre::eyre!("failed requesting block backfill: {err}")); + } + }; + backfill_job.lock().await.finish_send(request_id, &sent); + debug!(target: "zone::p2p", request_id, start, connected = sent.len(), requested = request_peers.len(), configured = backfill_peers.len(), "Sent block backfill request"); + } + + P2pCommand::SendBackfillBlock { + peer, + request_id, + block, + } => { + if block.len().saturating_add(9) > MAX_MESSAGE_SIZE as usize { + error!(target: "zone::p2p", block_size_bytes = block.len(), max_frame_size_bytes = MAX_MESSAGE_SIZE, "Backfill block exceeds the P2P response frame size limit"); + continue; } - debug!( - target: "zone::p2p", - "No followers are connected; retrying canonical block broadcast" - ); - tokio::time::sleep(BROADCAST_RETRY_INTERVAL).await; + let mut frame = Vec::with_capacity(block.len() + 9); + frame.push(BACKFILL_BLOCK_FRAME); + frame.extend_from_slice(&request_id.to_be_bytes()); + frame.extend_from_slice(&block); + senders + .backfill_responses + .send(Recipients::Some(vec![peer]), frame, true) + .await + .map_err(|err| eyre::eyre!("failed sending backfill block: {err}"))?; } - }) - .await; - let sent = match sent { - Ok(sent) => sent?, - Err(_) => { - warn!( - target: "zone::p2p", - timeout_secs = BROADCAST_RETRY_TIMEOUT.as_secs(), - "No followers connected before block broadcast timed out" - ); - continue; + + P2pCommand::CompleteBackfill { + peer, + request_id, + tip, + } => { + let mut frame = Vec::with_capacity(17); + frame.push(BACKFILL_COMPLETE_FRAME); + frame.extend_from_slice(&request_id.to_be_bytes()); + frame.extend_from_slice(&tip.to_be_bytes()); + senders + .backfill_responses + .send(Recipients::Some(vec![peer]), frame, true) + .await + .map_err(|err| eyre::eyre!("failed completing block backfill: {err}"))?; } - }; - if sent.len() != followers.len() { - debug!(target: "zone::p2p", connected = sent.len(), configured = followers.len(), "Some followers are not connected; block was not sent to them"); } } Err(eyre::eyre!("P2P command channel closed unexpectedly")) } -async fn run_block_receiver( +async fn run_receivers( role: Role, manifest: Arc, - mut receiver: lookup::Receiver, + receivers: P2pReceivers, + backfill_job: SharedBackfillLifecycle, events: mpsc::Sender, ) -> eyre::Result<()> { + let P2pReceivers { + mut blocks, + mut backfill_requests, + mut backfill_responses, + } = receivers; + let leader = manifest.leader_ed25519_public_key().clone(); loop { - let (peer, bytes) = receiver - .recv() - .await - .map_err(|err| eyre::eyre!("block channel receive failed: {err}"))?; - if role == Role::Leader { - warn!(target: "zone::p2p", %peer, "Leader received an unexpected block message"); - continue; - } - if peer != leader { - warn!(target: "zone::p2p", %peer, "Ignoring block from non-leader"); - continue; - } + let event = tokio::select! { + // Got a block + result = blocks.recv() => { + let (peer, bytes) = result.map_err(|err| eyre::eyre!("block channel receive failed: {err}"))?; + if role == Role::Leader || peer != leader { + warn!(target: "zone::p2p", %peer, "Ignoring live block from non-leader"); + continue; + } + P2pEvent::BlockReceived { leader_ed25519_public_key: peer, block: bytes.into() } + } + + // Got backfill request + result = backfill_requests.recv() => { + let (peer, bytes) = result.map_err(|err| eyre::eyre!("backfill request receive failed: {err}"))?; + let Ok(bytes): Result<[u8; 16], _> = bytes.as_ref().try_into() else { + warn!(target: "zone::p2p", %peer, size = bytes.len(), "Ignoring malformed backfill request"); + continue; + }; + let request_id = u64::from_be_bytes(bytes[..8].try_into().expect("fixed-size request ID")); + let start = u64::from_be_bytes(bytes[8..].try_into().expect("fixed-size backfill start")); + P2pEvent::BackfillRequested { peer, request_id, start } + } + + // Got backfill response (for an existing request) + result = backfill_responses.recv() => { + let (peer, bytes) = result.map_err(|err| eyre::eyre!("backfill response receive failed: {err}"))?; + let eligible = match role { Role::Leader => peer != leader, Role::Follower => peer == leader }; + if !eligible { + warn!(target: "zone::p2p", %peer, "Ignoring backfill response from ineligible peer"); + continue; + } + let Some((&frame_kind, frame_payload)) = bytes.as_ref().split_first() else { + warn!(target: "zone::p2p", %peer, "Ignoring empty backfill response frame"); + continue; + }; + let Some((request_id_bytes, payload)) = frame_payload.split_at_checked(8) else { + warn!(target: "zone::p2p", %peer, size = frame_payload.len(), "Ignoring backfill response without a request ID"); + continue; + }; + let request_id = u64::from_be_bytes(request_id_bytes.try_into().expect("fixed-size request ID")); + let received_at = Instant::now(); + + let mut backfill_job = backfill_job.lock().await; + match frame_kind { + BACKFILL_BLOCK_FRAME => { + if !backfill_job + .accepts(&peer, request_id, received_at) + { + warn!(target: "zone::p2p", %peer, request_id, "Ignoring unsolicited or stale backfill block"); + continue; + } + P2pEvent::BackfillBlockReceived { peer, block: payload.to_vec() } + } + BACKFILL_COMPLETE_FRAME => { + let accepted = backfill_job.complete(&peer, request_id, received_at); + if !accepted { + warn!(target: "zone::p2p", %peer, request_id, "Ignoring unsolicited or stale backfill completion"); + continue; + } + let Ok(tip_bytes): Result<[u8; 8], _> = payload.try_into() else { + warn!(target: "zone::p2p", %peer, request_id, size = payload.len(), "Ignoring malformed backfill completion"); + continue; + }; + P2pEvent::BackfillCompleted { peer, tip: u64::from_be_bytes(tip_bytes) } + } + _ => { + warn!(target: "zone::p2p", %peer, frame_kind, "Ignoring backfill response with unknown frame kind"); + continue; + } + } + } + }; events - .send(P2pEvent::BlockReceived { - leader_ed25519_public_key: peer, - block: bytes.into(), - }) + .send(event) .await .map_err(|_| eyre::eyre!("P2P event channel closed"))?; } @@ -399,14 +689,17 @@ mod tests { use std::{ net::{SocketAddr, TcpListener}, sync::Arc, - time::Duration, + time::{Duration, Instant}, }; use alloy_primitives::address; use commonware_codec::Encode as _; use commonware_cryptography::{Signer as _, ed25519::PrivateKey}; - use super::{P2pCommand, P2pConfig, P2pEvent, spawn_p2p, validate_ip_check_configuration}; + use super::{ + BACKFILL_RESPONSE_TIMEOUT, BackfillJob, P2pCommand, P2pConfig, P2pEvent, spawn_p2p, + validate_ip_check_configuration, + }; use crate::{P2pNetworkId, ZoneManifest, identity::Ed25519Identity, network::MAX_MESSAGE_SIZE}; fn available_address() -> SocketAddr { @@ -419,6 +712,66 @@ mod tests { Ed25519Identity::from_hex(&const_hex::encode_prefixed(key.encode().as_ref())).unwrap() } + #[test] + fn stale_response_cannot_complete_replacement_request() { + let peer = ed25519_identity(1).ed25519_public_key(); + let now = Instant::now(); + let mut lifecycle = BackfillJob::default(); + + let (first_id, peers) = lifecycle + .begin_request(std::slice::from_ref(&peer), now) + .unwrap(); + lifecycle.finish_send(first_id, &peers); + assert!(lifecycle.accepts(&peer, first_id, now)); + let halfway = now + BACKFILL_RESPONSE_TIMEOUT / 2; + assert!(lifecycle.accepts(&peer, first_id, halfway)); + assert!( + lifecycle + .begin_request(std::slice::from_ref(&peer), halfway) + .is_none() + ); + + let expired_at = now + BACKFILL_RESPONSE_TIMEOUT; + assert!(!lifecycle.accepts(&peer, first_id, expired_at)); + assert!(!lifecycle.complete(&peer, first_id, expired_at)); + + let (replacement_id, peers) = lifecycle + .begin_request(std::slice::from_ref(&peer), expired_at) + .unwrap(); + lifecycle.finish_send(replacement_id, &peers); + assert_ne!(first_id, replacement_id); + assert!(!lifecycle.accepts(&peer, first_id, expired_at)); + assert!(!lifecycle.complete(&peer, first_id, expired_at)); + assert!(lifecycle.accepts(&peer, replacement_id, expired_at)); + assert!(lifecycle.complete(&peer, replacement_id, expired_at)); + } + + async fn assert_no_backfill_response_events( + events: &mut tokio::sync::mpsc::Receiver, + duration: Duration, + context: &str, + ) { + let result = tokio::time::timeout(duration, async { + loop { + match events.recv().await { + Some(P2pEvent::BackfillBlockReceived { .. }) => { + panic!("{context}: accepted unsolicited backfill block") + } + Some(P2pEvent::BackfillCompleted { .. }) => { + panic!("{context}: accepted unsolicited backfill completion") + } + Some(_) => {} + None => return, + } + } + }) + .await; + assert!( + result.is_err(), + "{context}: event channel closed while checking for unsolicited backfill response" + ); + } + #[test] fn dns_manifest_requires_explicit_ip_check_bypass() { let identities = [ @@ -466,6 +819,8 @@ mod tests { )); } let manifest = Arc::new(ZoneManifest::parse(&input).unwrap()); + let leader_peer = identities[0].ed25519_public_key(); + let first_follower_peer = identities[1].ed25519_public_key(); let mut handles = identities .into_iter() .zip(addresses) @@ -522,6 +877,225 @@ mod tests { } broadcaster.abort(); + let leader_commands = handles[0].parts.as_ref().unwrap().commands.clone(); + let follower_commands = handles[1].parts.as_ref().unwrap().commands.clone(); + leader_commands + .send(P2pCommand::SendBackfillBlock { + peer: first_follower_peer.clone(), + request_id: 0, + block: block.clone(), + }) + .await + .unwrap(); + leader_commands + .send(P2pCommand::CompleteBackfill { + peer: first_follower_peer.clone(), + request_id: 0, + tip: 99, + }) + .await + .unwrap(); + assert_no_backfill_response_events( + handles[1].events_mut(), + Duration::from_secs(2), + "follower without an outstanding backfill request", + ) + .await; + + follower_commands + .send(P2pCommand::RequestBackfill { start: 7 }) + .await + .unwrap(); + let (requesting_peer, follower_request_id) = + tokio::time::timeout(Duration::from_secs(15), async { + loop { + if let Some(P2pEvent::BackfillRequested { + peer, + request_id, + start: 7, + }) = handles[0].events_mut().recv().await + { + return (peer, request_id); + } + } + }) + .await + .expect("leader did not receive backfill request"); + + follower_commands + .send(P2pCommand::RequestBackfill { start: 8 }) + .await + .unwrap(); + let duplicate_request = tokio::time::timeout(Duration::from_millis(750), async { + loop { + if let Some(P2pEvent::BackfillRequested { start: 8, .. }) = + handles[0].events_mut().recv().await + { + return; + } + } + }) + .await; + assert!( + duplicate_request.is_err(), + "follower sent a duplicate backfill request while the first response was outstanding" + ); + + follower_commands + .send(P2pCommand::SendBackfillBlock { + peer: leader_peer.clone(), + request_id: 0, + block: block.clone(), + }) + .await + .unwrap(); + follower_commands + .send(P2pCommand::CompleteBackfill { + peer: leader_peer.clone(), + request_id: 0, + tip: 100, + }) + .await + .unwrap(); + assert_no_backfill_response_events( + handles[0].events_mut(), + Duration::from_secs(2), + "leader without an outstanding backfill request", + ) + .await; + + leader_commands + .send(P2pCommand::SendBackfillBlock { + peer: requesting_peer.clone(), + request_id: follower_request_id, + block: block.clone(), + }) + .await + .unwrap(); + let second_block = vec![0xf8, 0x02, 0x80]; + leader_commands + .send(P2pCommand::SendBackfillBlock { + peer: requesting_peer.clone(), + request_id: follower_request_id, + block: second_block.clone(), + }) + .await + .unwrap(); + leader_commands + .send(P2pCommand::CompleteBackfill { + peer: requesting_peer, + request_id: follower_request_id, + tip: 9, + }) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(15), async { + let expected_blocks = [block.clone(), second_block]; + let mut received_blocks = 0; + loop { + match handles[1].events_mut().recv().await { + Some(P2pEvent::BackfillBlockReceived { block, .. }) => { + assert_eq!(block, expected_blocks[received_blocks]); + received_blocks += 1; + } + Some(P2pEvent::BackfillCompleted { tip: 9, .. }) => { + assert_eq!(received_blocks, expected_blocks.len()); + return; + } + Some(_) => {} + None => panic!("follower event channel closed"), + } + } + }) + .await + .expect("follower did not receive ordered backfill response"); + + leader_commands + .send(P2pCommand::SendBackfillBlock { + peer: first_follower_peer.clone(), + request_id: follower_request_id, + block: block.clone(), + }) + .await + .unwrap(); + leader_commands + .send(P2pCommand::CompleteBackfill { + peer: first_follower_peer.clone(), + request_id: follower_request_id, + tip: 10, + }) + .await + .unwrap(); + assert_no_backfill_response_events( + handles[1].events_mut(), + Duration::from_secs(2), + "follower after its backfill request completed", + ) + .await; + + leader_commands + .send(P2pCommand::RequestBackfill { start: 11 }) + .await + .unwrap(); + let leader_request_id = tokio::time::timeout(Duration::from_secs(15), async { + loop { + if let Some(P2pEvent::BackfillRequested { + request_id, + start: 11, + .. + }) = handles[1].events_mut().recv().await + { + return request_id; + } + } + }) + .await + .expect("follower did not receive recovering leader's backfill request"); + + follower_commands + .send(P2pCommand::SendBackfillBlock { + peer: leader_peer.clone(), + request_id: leader_request_id, + block: block.clone(), + }) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(15), async { + loop { + if let Some(P2pEvent::BackfillBlockReceived { + peer, + block: received, + }) = handles[0].events_mut().recv().await + { + assert!(peer == first_follower_peer); + assert_eq!(received, block); + return; + } + } + }) + .await + .expect("leader did not receive requested backfill block"); + follower_commands + .send(P2pCommand::CompleteBackfill { + peer: leader_peer, + request_id: leader_request_id, + tip: 12, + }) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(15), async { + loop { + if let Some(P2pEvent::BackfillCompleted { peer, tip: 12 }) = + handles[0].events_mut().recv().await + { + assert!(peer == first_follower_peer); + return; + } + } + }) + .await + .expect("leader did not receive requested backfill completion"); + for handle in handles { tokio::time::timeout(Duration::from_secs(10), handle.shutdown()) .await