Skip to content

refactor(app): route the duty path through ConsensusController #575

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
iamquang95 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

refactor(app): route the duty path through ConsensusController #575

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
60 changes: 33 additions & 27 deletions crates/app/src/node/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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}")]
Expand Down Expand Up @@ -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<OnceLock<Handle>>` pattern (see qbft::p2p `build_consensus_nodes`).
Expand All @@ -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.
Expand All @@ -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,
Expand Down Expand Up @@ -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,
Expand All @@ -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,
Expand Down Expand Up @@ -667,7 +672,7 @@ fn production_parsigex_seam(handles: &CoreHandles) -> ParSigExSeam {
)]
async fn run_lifecycle(
node: pluto_p2p::p2p::Node<CoreBehaviour>,
consensus: Arc<qbft::Consensus>,
consensus_controller: Arc<pluto_consensus::controller::ConsensusController>,
handles: CoreHandles,
wired: WiredComponents,
priv_key_lock: Option<Arc<privkeylock::Service>>,
Expand All @@ -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<Result<(), AppError>> = JoinSet::new();

Expand Down
56 changes: 30 additions & 26 deletions crates/app/src/node/wire.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -117,9 +117,9 @@ pub struct WireInputs {
pub submission_client: BeaconNodeClient,
/// Per-validator data for this node.
pub validators: Vec<ValidatorInfo>,
/// Already-constructed consensus component, also wired into the QBFT p2p
/// behaviour by the caller.
pub consensus: Arc<qbft::Consensus>,
/// Current consensus implementation, from the controller. Forwards to the
/// default QBFT impl the caller also wires into the QBFT p2p behaviour.
pub consensus: Arc<ConsensusWrapper>,
/// Whether the builder API is enabled.
pub builder_enabled: bool,
/// Upstream beacon URL the validator API reverse-proxies unhandled requests
Expand Down Expand Up @@ -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(())
Expand Down Expand Up @@ -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 ----
Expand Down Expand Up @@ -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
}
Expand Down Expand Up @@ -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<dyn std::error::Error + Send + Sync + 'static>),
}

/// Runs `run_consensus` for `duty`, bounded by the duty's deadline.
Expand All @@ -805,7 +809,7 @@ async fn run_bounded_by_duty_deadline<F, Fut>(
) -> Result<(), DutyConsensusError>
where
F: FnOnce(Duty, CancellationToken) -> Fut,
Fut: std::future::Future<Output = qbft::RunnerResult<()>>,
Fut: std::future::Future<Output = pluto_consensus::wrapper::Result<()>>,
{
let deadline = deadline_calc.deadline(&duty)?;
bounded_by_deadline(ct, deadline, move |dct| run_consensus(duty, dct)).await?;
Expand Down Expand Up @@ -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;
Expand Down
13 changes: 7 additions & 6 deletions crates/app/tests/wiring.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -210,7 +210,7 @@ fn wire_inputs(
eth2_cl: EthBeaconNodeApiClient,
beacon_client: BeaconNodeClient,
pubkey: PubKey,
consensus: Arc<qbft::Consensus>,
consensus: Arc<ConsensusWrapper>,
threshold: u64,
) -> WireInputs {
// Permissive verifier: the partial sigs carry arbitrary payloads, so real
Expand All @@ -234,7 +234,7 @@ fn wire_inputs_with(
eth2_cl: EthBeaconNodeApiClient,
beacon_client: BeaconNodeClient,
pubkey: PubKey,
consensus: Arc<qbft::Consensus>,
consensus: Arc<ConsensusWrapper>,
threshold: u64,
sigagg_verifier: VerifyFn,
) -> WireInputs {
Expand Down Expand Up @@ -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<qbft::Consensus> {
fn build_consensus(ct: &CancellationToken) -> Arc<ConsensusWrapper> {
let key = k256::SecretKey::random(&mut rand::thread_rng());
let (deadliner, expired_rx) = pluto_core::deadline::DeadlinerTask::start(
ct.clone(),
Expand All @@ -294,7 +294,7 @@ fn build_consensus(ct: &CancellationToken) -> Arc<qbft::Consensus> {
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,
Expand All @@ -309,7 +309,8 @@ fn build_consensus(ct: &CancellationToken) -> Arc<qbft::Consensus> {
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
Expand Down
40 changes: 28 additions & 12 deletions crates/consensus/src/controller.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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},
};
Expand Down Expand Up @@ -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.
Expand All @@ -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<qbft::Consensus>,
default_consensus: Arc<dyn Consensus>,
wrapped_consensus: ConsensusWrapper,
wrapped_consensus: Arc<ConsensusWrapper>,
}

impl ConsensusController {
Expand All @@ -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<dyn Consensus> = qbft;
let default_consensus: Arc<dyn Consensus> = Arc::clone(&qbft) as Arc<dyn Consensus>;

Ok(Self {
wrapped_consensus: ConsensusWrapper::new(default_consensus.clone()),
default_qbft: qbft,
wrapped_consensus: Arc::new(ConsensusWrapper::new(default_consensus.clone())),
default_consensus,
})
}
Expand All @@ -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<qbft::Consensus> {
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<ConsensusWrapper> {
Arc::clone(&self.wrapped_consensus)
}

/// Sets the current consensus implementation for `protocol`.
Expand Down Expand Up @@ -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::*;

Expand All @@ -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<dyn Consensus>),
&default_consensus
));

controller
.set_current_consensus_for_protocol(QBFT_V2_PROTOCOL_ID)
Expand All @@ -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,
Expand Down
Loading