diff --git a/Cargo.lock b/Cargo.lock index f82f4d0f..4fcb124a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5140,10 +5140,12 @@ dependencies = [ "pluto-eth2api", "pluto-eth2util", "pluto-featureset", + "pluto-infosync", "pluto-k1util", "pluto-p2p", "pluto-parsigex", "pluto-peerinfo", + "pluto-priority", "pluto-ssz", "pluto-testutil", "prost 0.14.4", diff --git a/crates/app/Cargo.toml b/crates/app/Cargo.toml index 1cd17e9e..fdbbf37f 100644 --- a/crates/app/Cargo.toml +++ b/crates/app/Cargo.toml @@ -22,6 +22,8 @@ pluto-eth2util.workspace = true pluto-p2p.workspace = true pluto-parsigex.workspace = true pluto-peerinfo.workspace = true +pluto-priority.workspace = true +pluto-infosync.workspace = true tokio.workspace = true tokio-util.workspace = true vise.workspace = true diff --git a/crates/app/src/node/behaviour.rs b/crates/app/src/node/behaviour.rs index e3d0eb28..6e3da8db 100644 --- a/crates/app/src/node/behaviour.rs +++ b/crates/app/src/node/behaviour.rs @@ -54,6 +54,8 @@ pub(crate) struct CoreBehaviour { pub consensus: qbft::p2p::Behaviour, /// Peer metadata exchange. pub peerinfo: peerinfo::Behaviour, + /// Priority protocol request/response transport (backs infosync). + pub priority: pluto_priority::p2p::Behaviour, } /// Async handles for driving the composed behaviour from the core workflow. @@ -65,28 +67,56 @@ pub struct CoreHandles { /// Shared P2P runtime context (known peers + live connections), used by the /// monitoring API's readiness checker to compute quorum connectivity. pub p2p_context: P2PContext, + /// Priority protocol component; started by the caller. + pub priority: Arc, + /// Expired-duty receiver for `priority`; move-only, consumed once by + /// `Component::start`. + pub priority_expired_rx: tokio::sync::mpsc::Receiver, + /// Infosync component driving the per-epoch priority exchange. + pub infosync: Arc, +} + +/// Inputs to [`wire_p2p`], grouped to keep the aggregating call site readable. +pub(crate) struct WireP2PParams { + pub key: k256::SecretKey, + pub p2p_config: pluto_p2p::config::P2PConfig, + pub peers: Vec, + pub consensus: Arc, + pub min_required: i64, + pub deadline_calc: Arc, + pub feature_set: Arc, + pub duty_gater: DutyGaterFn, + pub eth2_cl: EthBeaconNodeApiClient, + pub pub_shares_by_key: HashMap>, + pub lock_hash: Vec, + pub builder_enabled: bool, + pub nickname: String, + pub cancellation: CancellationToken, } /// Composes the core behaviours and builds the libp2p [`Node`]. // TODO(#402 part B): QUIC transport (featureset-gated off at v1.7.1) and // bandwidth metrics. -#[allow( - clippy::too_many_arguments, - reason = "wireP2P aggregates independent inputs; a config struct is deferred to part B when priority inputs are added" -)] pub(crate) async fn wire_p2p( - key: k256::SecretKey, - p2p_config: pluto_p2p::config::P2PConfig, - peers: Vec, - consensus: Arc, - duty_gater: DutyGaterFn, - eth2_cl: EthBeaconNodeApiClient, - pub_shares_by_key: HashMap>, - lock_hash: Vec, - builder_enabled: bool, - nickname: String, - cancellation: CancellationToken, + params: WireP2PParams, ) -> Result<(Node, CoreHandles), AppError> { + let WireP2PParams { + key, + p2p_config, + peers, + consensus, + min_required, + deadline_calc, + feature_set, + duty_gater, + eth2_cl, + pub_shares_by_key, + lock_hash, + builder_enabled, + nickname, + cancellation, + } = params; + let peer_ids = peers.iter().map(|peer| peer.id).collect::>(); let local_peer_id = peer::peer_id_from_key(key.public_key())?; @@ -123,6 +153,11 @@ pub(crate) async fn wire_p2p( ); let (parsigex_comp, parsigex_handle) = parsigex::Behaviour::new(parsigex_config); + // Priority rides the same QBFT consensus; clone before it moves into the + // QBFT behaviour below. + let priority_consensus: Arc = consensus.clone(); + let priority_cancellation = cancellation.clone(); + // QBFT consensus transport. `Behaviour::new` errors if the local peer id is // not present in the configured cluster peer list. let (consensus_comp, consensus_handle) = qbft::p2p::Behaviour::new(qbft::p2p::Config { @@ -146,6 +181,30 @@ pub(crate) async fn wire_p2p( .with_peers(peer_ids.clone()); let peerinfo_comp = peerinfo::Behaviour::new(local_peer_id, peerinfo_config); + // Priority + infosync: per-epoch negotiation of supported + // versions/protocols/proposal types. The 6s exchange timeout (half a slot) + // matches Charon; `new_component` fails fast on a peer missing from the + // shared `p2p_context`. + let (priority_comp, priority_behaviour, priority_expired_rx) = pluto_priority::new_component( + peer_ids.clone(), + min_required, + priority_consensus, + std::time::Duration::from_secs(6), + key.clone(), + deadline_calc, + p2p_context.clone(), + priority_cancellation, + )?; + let priority_comp = Arc::new(priority_comp); + + let infosync = Arc::new(pluto_infosync::Component::new( + Arc::clone(&priority_comp), + pluto_core::version::SUPPORTED.to_vec(), + local_protocols(), + local_proposal_types(builder_enabled), + &feature_set, + )); + // Clone the context before it is moved into the node so the readiness // checker observes the same shared peer/connection state the swarm updates. let p2p_context_for_handle = p2p_context.clone(); @@ -164,6 +223,7 @@ pub(crate) async fn wire_p2p( parsigex: parsigex_comp, consensus: consensus_comp, peerinfo: peerinfo_comp, + priority: priority_behaviour, }) }, )?; @@ -172,7 +232,72 @@ pub(crate) async fn wire_p2p( parsigex: parsigex_handle, consensus: consensus_handle, p2p_context: p2p_context_for_handle, + priority: priority_comp, + priority_expired_rx, + infosync, }; Ok((node, handles)) } + +/// Advertised proposal types in precedence order: builder first when enabled, +/// full always last as the fallback. +fn local_proposal_types(builder_enabled: bool) -> Vec { + let mut proposal_types = Vec::new(); + if builder_enabled { + proposal_types.push(pluto_core::types::ProposalType::Builder); + } + proposal_types.push(pluto_core::types::ProposalType::Full); + proposal_types +} + +/// Advertised protocols in precedence order: consensus, parsigex, peerinfo, +/// priority. +// TODO(#402 part B): reorder by the cluster-preferred / CLI consensus protocol +// once those inputs exist, matching Go's `PrioritizeProtocolsByName`. +fn local_protocols() -> Vec { + pluto_consensus::protocols::protocols() + .iter() + .map(|p| p.to_string()) + .chain(pluto_parsigex::protocols().iter().map(|p| p.to_string())) + .chain(pluto_peerinfo::protocols().iter().map(|p| p.to_string())) + .chain(pluto_priority::protocols().iter().map(|p| p.to_string())) + .collect() +} + +#[cfg(test)] +mod tests { + use pluto_core::types::ProposalType; + + use super::{local_proposal_types, local_protocols}; + + #[test] + fn proposal_types_put_builder_first_only_when_enabled() { + assert_eq!(local_proposal_types(false), vec![ProposalType::Full]); + assert_eq!( + local_proposal_types(true), + vec![ProposalType::Builder, ProposalType::Full], + ); + } + + #[test] + fn protocols_are_advertised_in_component_precedence_order() { + // This ordering is what keeps pluto's info_sync result byte-identical + // to Charon's; priority's absence is the bug this wiring fixes. + let got = local_protocols(); + + let index_of = |head: String| { + got.iter() + .position(|p| *p == head) + .expect("protocol advertised") + }; + let consensus = index_of(pluto_consensus::protocols::protocols()[0].to_string()); + let parsigex = index_of(pluto_parsigex::protocols()[0].to_string()); + let peerinfo = index_of(pluto_peerinfo::protocols()[0].to_string()); + let priority = index_of(pluto_priority::protocols()[0].to_string()); + + assert!(consensus < parsigex); + assert!(parsigex < peerinfo); + assert!(peerinfo < priority); + } +} diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 7a973b44..0a8a876c 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -123,6 +123,10 @@ pub enum AppError { #[error("consensus p2p: {0}")] ConsensusP2P(#[from] qbft::p2p::Error), + /// Priority protocol component construction failed. + #[error("priority: {0}")] + Priority(#[from] pluto_priority::Error), + /// A beacon node API request failed. #[error("beacon node api: {0}")] BeaconApi(#[from] pluto_eth2api::EthBeaconNodeApiClientError), @@ -377,8 +381,8 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { .map_err(AppError::Gater)? .into_fn(); - // Per-component deadline calculator, shared as an `Arc` so a single - // beacon-derived instance backs every component's deadliner. + // Shared `Arc` so one beacon-derived instance backs every + // component's deadliner (priority included). let deadline_calc: Arc = Arc::new( pluto_core::deadline::DutyDeadlineCalculator::from_client(ð2_cl) .await @@ -460,19 +464,24 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { let pub_shares_by_key = build_pub_shares_by_key(&lock)?; // ---- P2P behaviours (relay + parsigex + qbft + peerinfo) ---- - let (node, handles) = behaviour::wire_p2p( - key.clone(), - config.p2p.clone(), + let (node, handles) = behaviour::wire_p2p(behaviour::WireP2PParams { + key: key.clone(), + p2p_config: config.p2p.clone(), peers, - Arc::clone(&consensus), - Arc::clone(&duty_gater), - eth2_cl.clone(), + consensus: Arc::clone(&consensus), + // Priority quorum = cluster signing threshold (Charon's + // `int(cluster.GetThreshold())`). + min_required: i64::try_from(threshold).unwrap_or(i64::MAX), + deadline_calc: Arc::clone(&deadline_calc), + feature_set: Arc::clone(&feature_set), + duty_gater: Arc::clone(&duty_gater), + eth2_cl: eth2_cl.clone(), pub_shares_by_key, - lock.lock_hash.clone(), - config.builder_api, - config.nickname.clone(), - ct.clone(), - ) + lock_hash: lock.lock_hash.clone(), + builder_enabled: config.builder_api, + nickname: config.nickname.clone(), + cancellation: ct.clone(), + }) .await?; // Complete the broadcaster<->behaviour cycle. handle_slot @@ -547,6 +556,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { fetch_only_comm_idx0, seen_pubkeys: Some(seen_pubkeys_observer), slot_tick: vmock.clone().map(|v| simnet_slot_tick(v, ct.clone())), + infosync: Some(Arc::clone(&handles.infosync)), }, ct.clone(), ) @@ -689,6 +699,17 @@ async fn run_lifecycle( // Self-spawning actor: consensus expired-duty pruner. let _consensus_task = consensus.start(ct.clone()); + // Priority state-cleanup loop; the per-epoch infosync trigger is registered + // as a slot subscriber in `wire_core_workflow`. + handles + .priority + .start(handles.priority_expired_rx, ct.clone()); + + // TODO(#402 part B): consume the decided infosync result. Charon's + // `wirePrioritise` swaps the duty-consensus implementation via + // `ConsensusController::set_current_consensus_for_protocol`; deferred because + // it is a no-op while QBFTv2 is the only consensus protocol. + let mut tasks: JoinSet> = JoinSet::new(); // Supervise the scheduler actor alongside the other long-lived tasks so diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 697b2897..a5d5373f 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -153,6 +153,9 @@ pub struct WireInputs { /// Optional per-slot subscriber; simnet wires the in-process validator /// mock here. `None` in production and tests. pub slot_tick: Option, + /// Infosync component, triggered on each epoch's last slot to run the + /// cluster-wide priority exchange. `None` in tests. + pub infosync: Option>, } /// The wired components and long-lived handles produced by @@ -281,6 +284,7 @@ pub async fn wire_core_workflow( fetch_only_comm_idx0, seen_pubkeys, slot_tick, + infosync, } = inputs; // ---- Derived validator maps ---- @@ -608,6 +612,25 @@ pub async fn wire_core_workflow( if let Some(slot_tick) = slot_tick { sched_builder.subscribe_slot(move |slot: &Slot| slot_tick(slot), "simnet.vmock"); } + // Per-epoch infosync trigger, fired on each epoch's last slot. A failure is + // logged by `subscribe_slot` and does not fail the node. + if let Some(infosync) = infosync { + let ct = ct.clone(); + sched_builder.subscribe_slot( + move |slot: &Slot| { + let infosync = Arc::clone(&infosync); + let ct = ct.clone(); + let slot = slot.clone(); + async move { + if slot.last_in_epoch() { + infosync.trigger(ct.child_token(), slot.slot).await?; + } + Ok::<(), AppError>(()) + } + }, + "infosync", + ); + } // Slot subscriber: per-epoch validator cache trim + refresh (Charon's // `wireCoreWorkflow`). { diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs index 9580aa8b..da4b1637 100644 --- a/crates/app/tests/wiring.rs +++ b/crates/app/tests/wiring.rs @@ -269,6 +269,7 @@ fn wire_inputs_with( fetch_only_comm_idx0: false, seen_pubkeys: None, slot_tick: None, + infosync: None, } }