From 3981f7e9fda55eb50e21ceb474c3681d5fc06464 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Thu, 30 Jul 2026 12:25:13 +0000 Subject: [PATCH 1/3] feat(app): wire priority + infosync into the node Mount the priority protocol behaviour and construct the infosync component in the node's P2P wiring, and trigger a cluster-wide priority exchange on the last slot of each epoch. The priority component rides the existing QBFT consensus and shares the node-wide P2P context; infosync advertises this node's supported versions, protocols, and proposal types. This is the participation half of #402 part B ("A"): it makes pluto take part in the per-epoch priority/QBFT info_sync round so mixed clusters with Charon reach quorum. Consuming the decided result (routing the duty path through ConsensusController and registering the protocol-switch subscriber) is a functional no-op while QBFTv2 is the only consensus protocol and is left as a TODO(#402 part B). Verified on a 2-charon + 2-pluto kurtosis cluster: all four nodes reach identical info_sync results, and Charon's "protocols not supported: [charon/priority/2.0.0]" and priority "consensus timeout" warnings stop. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 2 + crates/app/Cargo.toml | 2 + crates/app/src/node/behaviour.rs | 69 ++++++++++++++++++++++++++++++++ crates/app/src/node/mod.rs | 39 +++++++++++++++++- crates/app/src/node/wire.rs | 27 +++++++++++++ crates/app/tests/wiring.rs | 1 + 6 files changed, 139 insertions(+), 1 deletion(-) 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..bb446b2a 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,6 +67,14 @@ 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. Held so the caller can start it and hand it + /// to the per-epoch infosync trigger. + pub priority: Arc, + /// Expired-duty receiver paired with `priority`, drained by + /// `pluto_priority::Component::start`. Taken (moved) once at start. + pub priority_expired_rx: tokio::sync::mpsc::Receiver, + /// Infosync component driving the per-epoch cluster-wide priority exchange. + pub infosync: Arc, } /// Composes the core behaviours and builds the libp2p [`Node`]. @@ -79,6 +89,9 @@ pub(crate) async fn wire_p2p( p2p_config: pluto_p2p::config::P2PConfig, peers: Vec, consensus: Arc, + min_required: i64, + deadline_calc: pluto_core::deadline::DutyDeadlineCalculator, + feature_set: Arc, duty_gater: DutyGaterFn, eth2_cl: EthBeaconNodeApiClient, pub_shares_by_key: HashMap>, @@ -123,6 +136,12 @@ pub(crate) async fn wire_p2p( ); let (parsigex_comp, parsigex_handle) = parsigex::Behaviour::new(parsigex_config); + // Priority protocol rides the same QBFT consensus instance and shares the + // node-wide `p2p_context`; clone both before they move 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 +165,52 @@ pub(crate) async fn wire_p2p( .with_peers(peer_ids.clone()); let peerinfo_comp = peerinfo::Behaviour::new(local_peer_id, peerinfo_config); + // Priority protocol + infosync: cluster-wide, per-epoch negotiation of the + // supported versions/protocols/proposal types, run over the shared QBFT + // consensus. `new_component` fails fast if any peer is absent from the + // shared `p2p_context`. The 6s exchange timeout (half a slot) matches the + // reference implementation. + 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); + + // Local proposal types in precedence order (builder first when enabled, + // full always last as the fallback). + 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); + + // Local supported protocols advertised on the infosync "protocol" topic, in + // precedence order: consensus, then parsigex, peerinfo, priority. + // TODO(#402 part B): reorder by the cluster-preferred / CLI consensus + // protocol once those inputs exist (pluto has neither yet), matching Go's + // `PrioritizeProtocolsByName` passes. + let 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(); + + let infosync = Arc::new(pluto_infosync::Component::new( + Arc::clone(&priority_comp), + pluto_core::version::SUPPORTED.to_vec(), + local_protocols, + proposal_types, + &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 +229,7 @@ pub(crate) async fn wire_p2p( parsigex: parsigex_comp, consensus: consensus_comp, peerinfo: peerinfo_comp, + priority: priority_behaviour, }) }, )?; @@ -172,6 +238,9 @@ 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)) diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 7a973b44..9ae786d6 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), @@ -378,12 +382,21 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { .into_fn(); // Per-component deadline calculator, shared as an `Arc` so a single - // beacon-derived instance backs every component's deadliner. + // beacon-derived instance backs every component's deadliner. The concrete + // instance is kept so the priority protocol (which needs an owned + // `DutyDeadlineCalculator`) can be given a clone of the same config. let deadline_calc: Arc = Arc::new( pluto_core::deadline::DutyDeadlineCalculator::from_client(ð2_cl) .await .map_err(AppError::Deadline)?, ); + // Priority needs an owned `DutyDeadlineCalculator` (it is not `Clone`), so + // build a second instance from the same client. The calculator is stateless + // beacon-derived config, so this is identical to the one above. + let priority_deadline_calc = + pluto_core::deadline::DutyDeadlineCalculator::from_client(ð2_cl) + .await + .map_err(AppError::Deadline)?; // Per-validator graffiti for proposed blocks. let graffiti_pubkeys: Vec = @@ -465,6 +478,12 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { config.p2p.clone(), peers, Arc::clone(&consensus), + // Priority's `min_required` is the cluster signing threshold (Charon's + // `int(cluster.GetThreshold())`): a priority survives only if proposed + // by at least this many peers. + i64::try_from(threshold).unwrap_or(i64::MAX), + priority_deadline_calc, + Arc::clone(&feature_set), Arc::clone(&duty_gater), eth2_cl.clone(), pub_shares_by_key, @@ -547,6 +566,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 +709,23 @@ async fn run_lifecycle( // Self-spawning actor: consensus expired-duty pruner. let _consensus_task = consensus.start(ct.clone()); + // Priority protocol state-cleanup loop, driven by its deadliner's + // expired-duty receiver. The receiver is move-only, so this runs once. + // The per-epoch infosync trigger that proposes into this component is + // registered as a scheduler 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` registers a second priority subscriber that reads the + // agreed "protocol" topic and calls + // `ConsensusController::set_current_consensus_for_protocol` to swap the + // duty-consensus implementation. Wiring that requires routing the duty path + // through a `ConsensusController` (today it is the raw QBFT consensus); it + // is deferred because the switch is a functional 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..8ecbc128 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -153,6 +153,10 @@ pub struct WireInputs { /// Optional per-slot subscriber; simnet wires the in-process validator /// mock here. `None` in production and tests. pub slot_tick: Option, + /// Optional infosync component. When `Some`, it is triggered on the last + /// slot of each epoch to run the cluster-wide priority exchange (supported + /// versions/protocols/proposal types) for the next epoch. `None` in tests. + pub infosync: Option>, } /// The wired components and long-lived handles produced by @@ -281,6 +285,7 @@ pub async fn wire_core_workflow( fetch_only_comm_idx0, seen_pubkeys, slot_tick, + infosync, } = inputs; // ---- Derived validator maps ---- @@ -608,6 +613,28 @@ 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: on the last slot of each epoch, run the + // cluster-wide priority exchange for the next epoch. A trigger failure is + // logged and swallowed — a missed info_sync must 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() + && let Err(err) = infosync.trigger(ct.child_token(), slot.slot).await + { + tracing::warn!(%err, slot = ?slot.slot, "infosync trigger failed"); + } + 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, } } From 669fe0fe91e14dd3830bfa7944447468a5510a08 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Fri, 31 Jul 2026 11:33:50 +0200 Subject: [PATCH 2/3] =?UTF-8?q?refactor(app):=20address=20#574=20review=20?= =?UTF-8?q?=E2=80=94=20params=20struct,=20shared=20deadliner,=20wiring=20t?= =?UTF-8?q?ests?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - wire_p2p: replace the 14 positional args (and its too_many_arguments allow) with a WireP2PParams struct destructured at the top. - Reuse the shared Arc for priority instead of building a second DutyDeadlineCalculator; wire_p2p now takes the Arc. - infosync slot subscriber: propagate the trigger error via `?` (the scheduler's subscribe_slot already logs it) instead of catching/warning. - Extract local_protocols()/local_proposal_types() as pure helpers and unit-test the Charon-parity protocol precedence and builder-first proposal ordering. Co-Authored-By: Claude Opus 4.8 --- crates/app/src/node/behaviour.rs | 147 ++++++++++++++++++++++--------- crates/app/src/node/mod.rs | 40 ++++----- crates/app/src/node/wire.rs | 8 +- 3 files changed, 125 insertions(+), 70 deletions(-) diff --git a/crates/app/src/node/behaviour.rs b/crates/app/src/node/behaviour.rs index bb446b2a..6cd306c7 100644 --- a/crates/app/src/node/behaviour.rs +++ b/crates/app/src/node/behaviour.rs @@ -77,29 +77,47 @@ pub struct CoreHandles { 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, - min_required: i64, - deadline_calc: pluto_core::deadline::DutyDeadlineCalculator, - feature_set: 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())?; @@ -182,32 +200,11 @@ pub(crate) async fn wire_p2p( )?; let priority_comp = Arc::new(priority_comp); - // Local proposal types in precedence order (builder first when enabled, - // full always last as the fallback). - 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); - - // Local supported protocols advertised on the infosync "protocol" topic, in - // precedence order: consensus, then parsigex, peerinfo, priority. - // TODO(#402 part B): reorder by the cluster-preferred / CLI consensus - // protocol once those inputs exist (pluto has neither yet), matching Go's - // `PrioritizeProtocolsByName` passes. - let 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(); - let infosync = Arc::new(pluto_infosync::Component::new( Arc::clone(&priority_comp), pluto_core::version::SUPPORTED.to_vec(), - local_protocols, - proposal_types, + local_protocols(), + local_proposal_types(builder_enabled), &feature_set, )); @@ -245,3 +242,71 @@ pub(crate) async fn wire_p2p( Ok((node, handles)) } + +/// Local proposal types advertised on the infosync "proposal" topic, 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 +} + +/// Local supported protocols advertised on the infosync "protocol" topic, in +/// precedence order: consensus, then parsigex, peerinfo, priority. +// TODO(#402 part B): reorder by the cluster-preferred / CLI consensus protocol +// once those inputs exist (pluto has neither yet), matching Go's +// `PrioritizeProtocolsByName` passes. +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() { + // The advertised order (consensus, then parsigex, peerinfo, priority) is + // what makes pluto's info_sync result byte-identical to Charon's. + let got = local_protocols(); + + // Every component's protocols must be present, priority included — its + // absence is exactly the bug this wiring fixes (Charon otherwise logs + // "protocols not supported: [charon/priority/2.0.0]"). + 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()); + + // Precedence: consensus < parsigex < peerinfo < priority. + 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 9ae786d6..064e90a5 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -382,21 +382,13 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { .into_fn(); // Per-component deadline calculator, shared as an `Arc` so a single - // beacon-derived instance backs every component's deadliner. The concrete - // instance is kept so the priority protocol (which needs an owned - // `DutyDeadlineCalculator`) can be given a clone of the same config. + // beacon-derived instance backs every component's deadliner (priority + // included — it takes a clone of this same `Arc`). let deadline_calc: Arc = Arc::new( pluto_core::deadline::DutyDeadlineCalculator::from_client(ð2_cl) .await .map_err(AppError::Deadline)?, ); - // Priority needs an owned `DutyDeadlineCalculator` (it is not `Clone`), so - // build a second instance from the same client. The calculator is stateless - // beacon-derived config, so this is identical to the one above. - let priority_deadline_calc = - pluto_core::deadline::DutyDeadlineCalculator::from_client(ð2_cl) - .await - .map_err(AppError::Deadline)?; // Per-validator graffiti for proposed blocks. let graffiti_pubkeys: Vec = @@ -473,25 +465,25 @@ 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), + consensus: Arc::clone(&consensus), // Priority's `min_required` is the cluster signing threshold (Charon's // `int(cluster.GetThreshold())`): a priority survives only if proposed // by at least this many peers. - i64::try_from(threshold).unwrap_or(i64::MAX), - priority_deadline_calc, - Arc::clone(&feature_set), - Arc::clone(&duty_gater), - eth2_cl.clone(), + 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 diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 8ecbc128..64e4427d 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -615,7 +615,7 @@ pub async fn wire_core_workflow( } // Per-epoch infosync trigger: on the last slot of each epoch, run the // cluster-wide priority exchange for the next epoch. A trigger failure is - // logged and swallowed — a missed info_sync must not fail the node. + // logged by `subscribe_slot` and does not fail the node. if let Some(infosync) = infosync { let ct = ct.clone(); sched_builder.subscribe_slot( @@ -624,10 +624,8 @@ pub async fn wire_core_workflow( let ct = ct.clone(); let slot = slot.clone(); async move { - if slot.last_in_epoch() - && let Err(err) = infosync.trigger(ct.child_token(), slot.slot).await - { - tracing::warn!(%err, slot = ?slot.slot, "infosync trigger failed"); + if slot.last_in_epoch() { + infosync.trigger(ct.child_token(), slot.slot).await?; } Ok::<(), AppError>(()) } From 75a61d1ac636cf99015e723af40ea28e34adc646 Mon Sep 17 00:00:00 2001 From: Bohdan Ohorodnii <273991985+varex83agent@users.noreply.github.com> Date: Fri, 31 Jul 2026 11:42:44 +0200 Subject: [PATCH 3/3] docs(app): trim verbose comments in the priority/infosync wiring Condense the wiring comments to the rationale that isn't obvious from the code (Charon parity, 6s exchange timeout, move-only expired receiver, deferred consensus-switch), dropping restatements of the code itself. Co-Authored-By: Claude Opus 4.8 --- crates/app/src/node/behaviour.rs | 43 +++++++++++++------------------- crates/app/src/node/mod.rs | 26 +++++++------------ crates/app/src/node/wire.rs | 8 +++--- 3 files changed, 29 insertions(+), 48 deletions(-) diff --git a/crates/app/src/node/behaviour.rs b/crates/app/src/node/behaviour.rs index 6cd306c7..6e3da8db 100644 --- a/crates/app/src/node/behaviour.rs +++ b/crates/app/src/node/behaviour.rs @@ -67,13 +67,12 @@ 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. Held so the caller can start it and hand it - /// to the per-epoch infosync trigger. + /// Priority protocol component; started by the caller. pub priority: Arc, - /// Expired-duty receiver paired with `priority`, drained by - /// `pluto_priority::Component::start`. Taken (moved) once at start. + /// 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 cluster-wide priority exchange. + /// Infosync component driving the per-epoch priority exchange. pub infosync: Arc, } @@ -154,9 +153,8 @@ pub(crate) async fn wire_p2p( ); let (parsigex_comp, parsigex_handle) = parsigex::Behaviour::new(parsigex_config); - // Priority protocol rides the same QBFT consensus instance and shares the - // node-wide `p2p_context`; clone both before they move into the QBFT - // behaviour below. + // 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(); @@ -183,11 +181,10 @@ pub(crate) async fn wire_p2p( .with_peers(peer_ids.clone()); let peerinfo_comp = peerinfo::Behaviour::new(local_peer_id, peerinfo_config); - // Priority protocol + infosync: cluster-wide, per-epoch negotiation of the - // supported versions/protocols/proposal types, run over the shared QBFT - // consensus. `new_component` fails fast if any peer is absent from the - // shared `p2p_context`. The 6s exchange timeout (half a slot) matches the - // reference implementation. + // 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, @@ -243,9 +240,8 @@ pub(crate) async fn wire_p2p( Ok((node, handles)) } -/// Local proposal types advertised on the infosync "proposal" topic, in -/// precedence order: builder first when enabled, full always last as the -/// fallback. +/// 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 { @@ -255,11 +251,10 @@ fn local_proposal_types(builder_enabled: bool) -> Vec Vec { pluto_consensus::protocols::protocols() .iter() @@ -287,13 +282,10 @@ mod tests { #[test] fn protocols_are_advertised_in_component_precedence_order() { - // The advertised order (consensus, then parsigex, peerinfo, priority) is - // what makes pluto's info_sync result byte-identical to Charon's. + // 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(); - // Every component's protocols must be present, priority included — its - // absence is exactly the bug this wiring fixes (Charon otherwise logs - // "protocols not supported: [charon/priority/2.0.0]"). let index_of = |head: String| { got.iter() .position(|p| *p == head) @@ -304,7 +296,6 @@ mod tests { let peerinfo = index_of(pluto_peerinfo::protocols()[0].to_string()); let priority = index_of(pluto_priority::protocols()[0].to_string()); - // Precedence: consensus < parsigex < peerinfo < priority. 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 064e90a5..0a8a876c 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -381,9 +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 (priority - // included — it takes a clone of this same `Arc`). + // 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 @@ -470,9 +469,8 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { p2p_config: config.p2p.clone(), peers, consensus: Arc::clone(&consensus), - // Priority's `min_required` is the cluster signing threshold (Charon's - // `int(cluster.GetThreshold())`): a priority survives only if proposed - // by at least this many peers. + // 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), @@ -701,22 +699,16 @@ async fn run_lifecycle( // Self-spawning actor: consensus expired-duty pruner. let _consensus_task = consensus.start(ct.clone()); - // Priority protocol state-cleanup loop, driven by its deadliner's - // expired-duty receiver. The receiver is move-only, so this runs once. - // The per-epoch infosync trigger that proposes into this component is - // registered as a scheduler slot subscriber in `wire_core_workflow`. + // 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` registers a second priority subscriber that reads the - // agreed "protocol" topic and calls - // `ConsensusController::set_current_consensus_for_protocol` to swap the - // duty-consensus implementation. Wiring that requires routing the duty path - // through a `ConsensusController` (today it is the raw QBFT consensus); it - // is deferred because the switch is a functional no-op while QBFTv2 is the - // only consensus protocol. + // `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(); diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 64e4427d..a5d5373f 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -153,9 +153,8 @@ pub struct WireInputs { /// Optional per-slot subscriber; simnet wires the in-process validator /// mock here. `None` in production and tests. pub slot_tick: Option, - /// Optional infosync component. When `Some`, it is triggered on the last - /// slot of each epoch to run the cluster-wide priority exchange (supported - /// versions/protocols/proposal types) for the next epoch. `None` in tests. + /// Infosync component, triggered on each epoch's last slot to run the + /// cluster-wide priority exchange. `None` in tests. pub infosync: Option>, } @@ -613,8 +612,7 @@ 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: on the last slot of each epoch, run the - // cluster-wide priority exchange for the next epoch. A trigger failure is + // 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();