From 42204897b798433a24202265412dc8ddbd968cb3 Mon Sep 17 00:00:00 2001 From: Quang Le Date: Fri, 31 Jul 2026 14:59:23 +0700 Subject: [PATCH] refactor(app): route the duty path through ConsensusController --- crates/app/src/node/mod.rs | 60 ++++++++++++++++-------------- crates/app/src/node/wire.rs | 56 +++++++++++++++------------- crates/app/tests/wiring.rs | 13 ++++--- crates/consensus/src/controller.rs | 40 ++++++++++++++------ 4 files changed, 98 insertions(+), 71 deletions(-) diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 7a973b44..1ce3a3be 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -115,9 +115,9 @@ pub enum AppError { #[error("relays: {0}")] Relays(#[from] pluto_p2p::bootnode::BootnodeError), - /// QBFT consensus construction failed. - #[error("consensus: {0}")] - Consensus(#[from] qbft::Error), + /// Consensus controller construction failed. + #[error("consensus controller: {0}")] + ConsensusController(#[from] pluto_consensus::controller::Error), /// QBFT p2p adapter construction failed. #[error("consensus p2p: {0}")] @@ -410,10 +410,10 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { let fetch_only_comm_idx0 = feature_set.enabled(pluto_featureset::Feature::FetchOnlyCommIdx0); - // ---- Consensus (built directly; shared with p2p behaviour + core stitch) ---- + // ---- Consensus (controller-owned) ---- // - // TODO(#402 part B): wrap in ConsensusController for dynamic protocol - // switching (priority/infosync). + // TODO(#402 part B): drive `set_current_consensus_for_protocol` from + // `prio.subscribe` once priority/infosync is wired. // // Resolve the broadcaster<->behaviour construction cycle with the // `Arc>` pattern (see qbft::p2p `build_consensus_nodes`). @@ -439,21 +439,26 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { }) }; - let consensus = Arc::new(qbft::Consensus::new(qbft::Config { - peers: qbft_peers, - local_peer_idx: i64::try_from(local_idx).map_err(|_| AppError::LocalPeerNotFound)?, - privkey: key.clone(), - deadliner: cons_deadliner, - expired_rx: cons_expired_rx, - duty_gater: Arc::clone(&duty_gater), - broadcaster, - sniffer: Arc::new(|_| {}), - // Charon gates duplicate-attestation comparison on the alpha - // `ChainSplitHalt` featureset flag (off by default). - compare_attestations: feature_set.enabled(pluto_featureset::Feature::ChainSplitHalt), - feature_set: Arc::clone(&feature_set), - timer_func: pluto_consensus::timer::get_round_timer_func(Arc::clone(&feature_set)), - })?); + // The controller owns the default QBFT impl and the swappable wrapper the + // duty path runs through. QBFTv2 is the only protocol today, so no swap + // ever happens; the indirection is what a second protocol would hook into. + let consensus_controller = Arc::new(pluto_consensus::controller::ConsensusController::new( + pluto_consensus::controller::Config { + peers: qbft_peers, + local_peer_idx: i64::try_from(local_idx).map_err(|_| AppError::LocalPeerNotFound)?, + privkey: key.clone(), + deadliner: cons_deadliner, + expired_rx: cons_expired_rx, + duty_gater: Arc::clone(&duty_gater), + broadcaster, + sniffer: Arc::new(|_| {}), + // Charon gates duplicate-attestation comparison on the alpha + // `ChainSplitHalt` featureset flag (off by default). + compare_attestations: feature_set.enabled(pluto_featureset::Feature::ChainSplitHalt), + feature_set: Arc::clone(&feature_set), + timer_func: pluto_consensus::timer::get_round_timer_func(Arc::clone(&feature_set)), + }, + )?); // Full public-share map (DV root pubkey -> share index -> public share), // used by the parsigex verifier to check each peer's partial signature. @@ -464,7 +469,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { key.clone(), config.p2p.clone(), peers, - Arc::clone(&consensus), + consensus_controller.default_qbft(), Arc::clone(&duty_gater), eth2_cl.clone(), pub_shares_by_key, @@ -536,7 +541,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { eth2_cl, submission_client, validators, - consensus: Arc::clone(&consensus), + consensus: consensus_controller.current_consensus(), builder_enabled: config.builder_api, upstream_url, parsigex: parsigex_seam, @@ -555,7 +560,7 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { // ---- Lifecycle: spawn long-lived tasks ---- run_lifecycle( node, - consensus, + consensus_controller, handles, wired, priv_key_lock, @@ -667,7 +672,7 @@ fn production_parsigex_seam(handles: &CoreHandles) -> ParSigExSeam { )] async fn run_lifecycle( node: pluto_p2p::p2p::Node, - consensus: Arc, + consensus_controller: Arc, handles: CoreHandles, wired: WiredComponents, priv_key_lock: Option>, @@ -686,8 +691,9 @@ async fn run_lifecycle( validator_api_router, } = wired; - // Self-spawning actor: consensus expired-duty pruner. - let _consensus_task = consensus.start(ct.clone()); + // Self-spawning actor: consensus expired-duty pruner. Starts the default + // impl only; a swapped-in protocol would be started by the controller. + consensus_controller.start(ct.clone()); let mut tasks: JoinSet> = JoinSet::new(); diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 697b2897..d7a216da 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -19,7 +19,7 @@ use std::{collections::HashMap, sync::Arc}; use futures::future::BoxFuture; -use pluto_consensus::qbft; +use pluto_consensus::wrapper::ConsensusWrapper; use pluto_core::{ aggsigdb::{memory::MemoryDBHandle, types::AggSigDB}, bcast::Broadcaster, @@ -117,9 +117,9 @@ pub struct WireInputs { pub submission_client: BeaconNodeClient, /// Per-validator data for this node. pub validators: Vec, - /// Already-constructed consensus component, also wired into the QBFT p2p - /// behaviour by the caller. - pub consensus: Arc, + /// Current consensus implementation, from the controller. Forwards to the + /// default QBFT impl the caller also wires into the QBFT p2p behaviour. + pub consensus: Arc, /// Whether the builder API is enabled. pub builder_enabled: bool, /// Upstream beacon URL the validator API reverse-proxies unhandled requests @@ -376,7 +376,7 @@ pub async fn wire_core_workflow( &deadline_calc, &ct, duty, - move |duty, dct| async move { consensus.propose(duty, value, &dct).await }, + move |duty, dct| async move { consensus.propose(dct, duty, value).await }, ) .await?; Ok(()) @@ -404,23 +404,25 @@ pub async fn wire_core_workflow( // we spawn the async `dutydb.store` inside it. { let dutydb = Arc::clone(&dutydb); - consensus.subscribe(move |duty: Duty, value: pbcore::UnsignedDataSet| { - let dutydb = Arc::clone(&dutydb); - tokio::spawn(async move { - let core_set = - match unsigneddata::unsigned_data_set_from_proto(&duty.duty_type, &value) { - Ok(set) => set, - Err(err) => { - tracing::warn!(?err, "dutydb: decode unsigned data set"); - return; - } - }; - if let Err(err) = dutydb.store(duty, core_set).await { - tracing::warn!(?err, "dutydb: store"); - } - }); - Ok(()) - }); + consensus.subscribe(Box::new( + move |duty: Duty, value: pbcore::UnsignedDataSet| { + let dutydb = Arc::clone(&dutydb); + tokio::spawn(async move { + let core_set = + match unsigneddata::unsigned_data_set_from_proto(&duty.duty_type, &value) { + Ok(set) => set, + Err(err) => { + tracing::warn!(?err, "dutydb: decode unsigned data set"); + return; + } + }; + if let Err(err) = dutydb.store(duty, core_set).await { + tracing::warn!(?err, "dutydb: store"); + } + }); + Ok(()) + }, + )); } // ---- (8) ParSigDB ---- @@ -596,7 +598,7 @@ pub async fn wire_core_workflow( &deadline_calc, &ct, duty, - move |duty, dct| async move { consensus.participate(duty, &dct).await }, + move |duty, dct| async move { consensus.participate(dct, duty).await }, ) .await } @@ -788,8 +790,10 @@ where enum DutyConsensusError { #[error(transparent)] Deadline(#[from] pluto_core::deadline::DeadlineError), + // Boxed rather than `qbft::RunnerError`: the wrapper erases the concrete + // impl's error type so a swapped-in protocol can report its own. #[error(transparent)] - Consensus(#[from] qbft::RunnerError), + Consensus(#[from] Box), } /// Runs `run_consensus` for `duty`, bounded by the duty's deadline. @@ -805,7 +809,7 @@ async fn run_bounded_by_duty_deadline( ) -> Result<(), DutyConsensusError> where F: FnOnce(Duty, CancellationToken) -> Fut, - Fut: std::future::Future>, + Fut: std::future::Future>, { let deadline = deadline_calc.deadline(&duty)?; bounded_by_deadline(ct, deadline, move |dct| run_consensus(duty, dct)).await?; @@ -881,7 +885,7 @@ mod deadline_bound_tests { let ran_probe = Arc::clone(&ran_probe); async move { ran_probe.store(true, Ordering::SeqCst); - Ok::<(), qbft::RunnerError>(()) + Ok(()) } }) .await; diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs index 9580aa8b..de99af5b 100644 --- a/crates/app/tests/wiring.rs +++ b/crates/app/tests/wiring.rs @@ -35,7 +35,7 @@ use pluto_app::node::{ ParSigExReceived, ParSigExSeam, SlotTickFn, ValidatorInfo, WireInputs, wire_core_workflow, }, }; -use pluto_consensus::qbft; +use pluto_consensus::{qbft, wrapper::ConsensusWrapper}; use pluto_core::{ aggsigdb::types::AggSigDB, sigagg::VerifyFn, @@ -210,7 +210,7 @@ fn wire_inputs( eth2_cl: EthBeaconNodeApiClient, beacon_client: BeaconNodeClient, pubkey: PubKey, - consensus: Arc, + consensus: Arc, threshold: u64, ) -> WireInputs { // Permissive verifier: the partial sigs carry arbitrary payloads, so real @@ -234,7 +234,7 @@ fn wire_inputs_with( eth2_cl: EthBeaconNodeApiClient, beacon_client: BeaconNodeClient, pubkey: PubKey, - consensus: Arc, + consensus: Arc, threshold: u64, sigagg_verifier: VerifyFn, ) -> WireInputs { @@ -281,7 +281,7 @@ fn pubkey_to_eth2(pk: PubKey) -> phase0::BLSPubKey { /// Builds a minimal single-node QBFT consensus component (not driven; only used /// to satisfy the wiring — its subscribe/propose are wired but not exercised in /// this test). -fn build_consensus(ct: &CancellationToken) -> Arc { +fn build_consensus(ct: &CancellationToken) -> Arc { let key = k256::SecretKey::random(&mut rand::thread_rng()); let (deadliner, expired_rx) = pluto_core::deadline::DeadlinerTask::start( ct.clone(), @@ -294,7 +294,7 @@ fn build_consensus(ct: &CancellationToken) -> Arc { public_key: key.public_key(), }; let feature_set = Arc::new(pluto_featureset::FeatureSet::new()); - Arc::new( + let consensus = Arc::new( qbft::Consensus::new(qbft::Config { peers: vec![peer], local_peer_idx: 0, @@ -309,7 +309,8 @@ fn build_consensus(ct: &CancellationToken) -> Arc { timer_func: pluto_consensus::timer::get_round_timer_func(feature_set), }) .expect("consensus"), - ) + ); + Arc::new(ConsensusWrapper::new(consensus)) } /// (a) Fetcher → AggSigDB back-edge (proposer/RANDAO) and diff --git a/crates/consensus/src/controller.rs b/crates/consensus/src/controller.rs index db882311..6b61c9f2 100644 --- a/crates/consensus/src/controller.rs +++ b/crates/consensus/src/controller.rs @@ -9,8 +9,7 @@ use tokio::sync::mpsc; use tokio_util::sync::CancellationToken; use crate::{ - debugger::Debugger, - qbft, + qbft::{self, SnifferSink}, timer::RoundTimerFunc, wrapper::{Consensus, ConsensusWrapper}, }; @@ -46,8 +45,8 @@ pub struct Config { pub duty_gater: DutyGaterFn, /// External message broadcaster. pub broadcaster: qbft::Broadcaster, - /// Consensus debugger. - pub debugger: Debugger, + /// Completed sniffer sink. + pub sniffer: SnifferSink, /// Enables attestation value comparison. pub compare_attestations: bool, /// Round timer factory. @@ -58,8 +57,12 @@ pub struct Config { /// Controls the active consensus protocol implementation. pub struct ConsensusController { + /// Same instance as `default_consensus`, kept concrete because the QBFT p2p + /// behaviour and the priority protocol bind to the concrete type. Go has no + /// equivalent: `qbft.NewConsensus` registers its own stream handler. + default_qbft: Arc, default_consensus: Arc, - wrapped_consensus: ConsensusWrapper, + wrapped_consensus: Arc, } impl ConsensusController { @@ -73,15 +76,16 @@ impl ConsensusController { expired_rx: config.expired_rx, duty_gater: config.duty_gater, broadcaster: config.broadcaster, - sniffer: config.debugger.sniffer(), + sniffer: config.sniffer, compare_attestations: config.compare_attestations, timer_func: config.timer_func, feature_set: config.feature_set, })?); - let default_consensus: Arc = qbft; + let default_consensus: Arc = Arc::clone(&qbft) as Arc; Ok(Self { - wrapped_consensus: ConsensusWrapper::new(default_consensus.clone()), + default_qbft: qbft, + wrapped_consensus: Arc::new(ConsensusWrapper::new(default_consensus.clone())), default_consensus, }) } @@ -96,9 +100,15 @@ impl ConsensusController { Arc::clone(&self.default_consensus) } + /// Returns the concrete default QBFT implementation, for wiring the QBFT + /// p2p behaviour and the priority protocol. + pub fn default_qbft(&self) -> Arc { + Arc::clone(&self.default_qbft) + } + /// Returns the current consensus wrapper. - pub fn current_consensus(&self) -> &ConsensusWrapper { - &self.wrapped_consensus + pub fn current_consensus(&self) -> Arc { + Arc::clone(&self.wrapped_consensus) } /// Sets the current consensus implementation for `protocol`. @@ -131,7 +141,7 @@ mod tests { types::DutyType, }; - use crate::{debugger::Debugger, protocols::QBFT_V2_PROTOCOL_ID, timer::get_round_timer_func}; + use crate::{protocols::QBFT_V2_PROTOCOL_ID, timer::get_round_timer_func}; use super::*; @@ -148,6 +158,12 @@ mod tests { controller.current_consensus().protocol_id(), QBFT_V2_PROTOCOL_ID ); + // Must be one instance: otherwise p2p would feed a different impl than + // the one driving duties. + assert!(Arc::ptr_eq( + &(controller.default_qbft() as Arc), + &default_consensus + )); controller .set_current_consensus_for_protocol(QBFT_V2_PROTOCOL_ID) @@ -174,7 +190,7 @@ mod tests { expired_rx, duty_gater: Arc::new(|duty| duty.duty_type == DutyType::Attester), broadcaster: Arc::new(|_, _| Box::pin(async { Ok(()) })), - debugger: Debugger::new(), + sniffer: Arc::new(|_| {}), compare_attestations: false, timer_func: get_round_timer_func(fs.clone()), feature_set: fs,