diff --git a/crates/app/src/node/mod.rs b/crates/app/src/node/mod.rs index 7a973b44..685f2f52 100644 --- a/crates/app/src/node/mod.rs +++ b/crates/app/src/node/mod.rs @@ -262,6 +262,16 @@ async fn run(config: AppConfig, ct: CancellationToken) -> Result<(), AppError> { let peers = lock.peers()?; pluto_p2p::peer::verify_p2p_key(&peers, &key)?; + // Tracker view of the cluster, captured before `peers` moves into the P2P + // wiring. Share indices are 1-indexed, matching partial-signature share ids. + let tracker_peers: Vec = peers + .iter() + .map(|peer| pluto_core::tracker::PeerInfo { + name: peer.name.clone(), + share_idx: usize::try_from(peer.index.saturating_add(1)).unwrap_or(usize::MAX), + }) + .collect(); + // Cluster size + quorum for the health-checker metadata, captured before // `peers` is moved into the P2P wiring. let num_peers = i64::try_from(peers.len()).unwrap_or(i64::MAX); @@ -547,6 +557,8 @@ 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())), + peers: tracker_peers, + feature_set: Arc::clone(&feature_set), }, ct.clone(), ) @@ -683,6 +695,7 @@ async fn run_lifecycle( parsigdb_deadliner_rx, aggsigdb: _aggsigdb, fetcher: _fetcher, + inclusion_checker, validator_api_router, } = wired; @@ -716,6 +729,16 @@ async fn run_lifecycle( }); } + // Networked inclusion checker: polls the beacon node once per due slot and + // resolves each tracked duty's on-chain inclusion step. + { + let ct = ct.clone(); + tasks.spawn(async move { + inclusion_checker.run(ct).await; + Ok(()) + }); + } + // Private-key lock maintenance loop. Only spawn `run` when locking is // enabled — `Service::close()` blocks forever unless `run` was called, so // spawn and close are guarded by the same `Option`. diff --git a/crates/app/src/node/wire.rs b/crates/app/src/node/wire.rs index 697b2897..510dbae4 100644 --- a/crates/app/src/node/wire.rs +++ b/crates/app/src/node/wire.rs @@ -33,6 +33,10 @@ use pluto_core::{ scheduler::SchedulerBuilder, sigagg::{Aggregator, VerifyFn}, signeddata::{SyncContribution, VersionedAggregatedAttestation}, + tracker::{ + AnalyserRx, DeleterRx, PeerInfo, StepError, Tracker, TrackerService, + inclusion::{INCL_CHECK_LAG, INCL_MISSED_LAG, InclusionChecker}, + }, types::{Duty, ParSignedData, ParSignedDataSet, PubKey, SignedData, SignedDataSet, Slot}, unsigneddata::{self, UnsignedDataSet}, validatorapi::{self, Component, Handler, SeenPubkeysFn}, @@ -42,6 +46,7 @@ use pluto_eth2api::{ spec::{bellatrix::ExecutionAddress, phase0::BLSPubKey}, valcache::{ValidatorCache, ValidatorCacheError}, }; +use pluto_featureset::{Feature, FeatureSet, Status}; use tokio_util::sync::CancellationToken; use crate::node::AppError; @@ -98,6 +103,144 @@ pub struct ValidatorInfo { pub fee_recipient: ExecutionAddress, } +/// Shares one step error between the tracker and the caller that must still +/// propagate it. +/// +/// The tracker needs an owned [`StepError`] (`Arc`), but the stitch +/// points also have to return their original error, and those error types are +/// not `Clone`. Moving the error into an `Arc` and handing the caller this +/// wrapper keeps a single allocation while preserving the chain: `source()` +/// returns the original error, so the tracker's reason inference — which walks +/// `source()` looking for an `EthBeaconNodeApiClientError` — still classifies +/// beacon-node failures correctly. +#[derive(Debug, Clone)] +struct SharedStepError(StepError); + +impl std::fmt::Display for SharedStepError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + self.0.fmt(f) + } +} + +impl std::error::Error for SharedStepError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(&*self.0) + } +} + +/// Splits a step result into the error to report to the tracker and the error +/// to return to the caller, sharing one allocation between them. +fn share_step_err(err: E) -> (StepError, SharedStepError) +where + E: std::error::Error + Send + Sync + 'static, +{ + let shared: StepError = Arc::new(err); + (Arc::clone(&shared), SharedStepError(shared)) +} + +/// Reports a step error to the tracker without needing to propagate it, for +/// stitch points whose errors are logged and swallowed locally. +fn owned_step_err(err: E) -> StepError +where + E: std::error::Error + Send + Sync + 'static, +{ + Arc::new(err) +} + +/// Wraps a [`DeadlineCalculator`], shifting every deadline later by a fixed +/// offset. Used to derive the tracker's analyser/deleter deadlines from the +/// shared duty deadline. Parity: the closures charon builds in `newTracker`. +struct OffsetCalculator { + inner: Arc, + offset: std::time::Duration, +} + +impl OffsetCalculator { + fn new(inner: Arc, offset: std::time::Duration) -> Self { + Self { inner, offset } + } +} + +impl DeadlineCalculator for OffsetCalculator { + fn deadline( + &self, + duty: &Duty, + ) -> pluto_core::deadline::Result>> { + let Some(deadline) = self.inner.deadline(duty)? else { + return Ok(None); + }; + // A deadline that cannot be shifted (overflow) is treated as + // never-expiring rather than silently wrapping to an earlier instant. + let shifted = chrono::Duration::from_std(self.offset) + .ok() + .and_then(|offset| deadline.checked_add_signed(offset)); + Ok(shifted) + } +} + +/// Feature set the tracker subsystem runs under. +/// +/// The networked inclusion checker only resolves proposer inclusion; the +/// attestation-inclusion path (attester/aggregator) is a follow-up, and its +/// core panics if fed those submissions. So mask the (alpha, off-by-default) +/// `AttestationInclusion` feature off until that path lands, keeping the +/// analyser and the checker consistent. +fn tracker_feature_set(feature_set: &Arc) -> Arc { + if !feature_set.enabled(Feature::AttestationInclusion) { + return Arc::clone(feature_set); + } + + tracing::warn!( + "Feature attestation_inclusion is enabled but not yet supported by the \ + inclusion checker; disabling it for duty tracking" + ); + let mut fs = (**feature_set).clone(); + fs.state + .insert(Feature::AttestationInclusion, Status::Disable); + Arc::new(fs) +} + +/// Returns the slot to start tracking from, which suppresses noisy failed +/// duties at startup caused by a validator client that is still coming up. +/// +/// Delays at most 10 seconds but never fewer than 2 slots. Parity: charon +/// `app.go` `calculateTrackerDelay`. +async fn calculate_tracker_delay( + eth2_cl: &EthBeaconNodeApiClient, + slot_duration: std::time::Duration, +) -> Result { + const MAX_DELAY: std::time::Duration = std::time::Duration::from_secs(10); + const MIN_DELAY_SLOTS: u64 = 2; + + let genesis = eth2_cl + .fetch_genesis_time() + .await + .map_err(AppError::BeaconApi)?; + + let elapsed = chrono::Utc::now() + .signed_duration_since(genesis) + .to_std() + .unwrap_or(std::time::Duration::ZERO); + let slot_nanos = slot_duration.as_nanos(); + let current_slot = elapsed + .as_nanos() + .checked_div(slot_nanos) + .and_then(|slots| u64::try_from(slots).ok()) + .unwrap_or(u64::MAX); + + let max_delay_slots = MAX_DELAY + .as_nanos() + .checked_div(slot_nanos) + .and_then(|slots| u64::try_from(slots).ok()) + .unwrap_or(u64::MAX); + let max_delay_time_slot = current_slot + .saturating_add(max_delay_slots) + .saturating_add(1); + let min_delay_slot = current_slot.saturating_add(MIN_DELAY_SLOTS); + + Ok(max_delay_time_slot.max(min_delay_slot)) +} + /// Already-resolved inputs to [`wire_core_workflow`]. /// /// These are derived from the cluster manifest + config, but passed in so that @@ -153,6 +296,12 @@ pub struct WireInputs { /// Optional per-slot subscriber; simnet wires the in-process validator /// mock here. `None` in production and tests. pub slot_tick: Option, + /// Cluster peers, used by the tracker to attribute per-peer participation. + /// Empty disables participation reporting but still tracks duty outcomes. + pub peers: Vec, + /// Resolved feature set. The tracker consults it to decide which duty types + /// have an on-chain inclusion step (`Feature::AttestationInclusion`). + pub feature_set: Arc, } /// The wired components and long-lived handles produced by @@ -173,6 +322,9 @@ pub struct WiredComponents { pub aggsigdb: MemoryDBHandle, /// The fetcher (driven via scheduler subscriptions). pub fetcher: Arc, + /// Networked inclusion checker; its `run` loop is spawned and supervised by + /// the caller. + pub inclusion_checker: Arc, /// The validator API axum router, ready to be served. pub validator_api_router: axum::Router, } @@ -281,6 +433,8 @@ pub async fn wire_core_workflow( fetch_only_comm_idx0, seen_pubkeys, slot_tick, + peers, + feature_set, } = inputs; // ---- Derived validator maps ---- @@ -323,6 +477,75 @@ pub async fn wire_core_workflow( let (aggsigdb_deadliner, aggsigdb_deadliner_rx) = DeadlinerTask::start(ct.clone(), "aggsigdb", Arc::clone(&deadline_calc)); + // ---- Tracker ---- + // + // Analysis has to wait until a duty's inclusion verdict can have arrived, so + // both tracker deadliners sit `INCL_MISSED_LAG + INCL_CHECK_LAG` slots past + // the duty deadline, and the deleter a further minute past the analyser so + // duties of the same slot are analysed before their events are dropped. + // Parity: charon `app.go` `newTracker`. + let (slot_duration, _slots_per_epoch) = eth2_cl + .fetch_slots_config() + .await + .map_err(AppError::BeaconApi)?; + let tracker_lag = slot_duration + .saturating_mul(u32::try_from(INCL_MISSED_LAG + INCL_CHECK_LAG).unwrap_or(u32::MAX)); + + let (tracker_analyser, tracker_analyser_rx) = DeadlinerTask::start( + ct.clone(), + "tracker_analyser", + Arc::new(OffsetCalculator::new( + Arc::clone(&deadline_calc), + tracker_lag, + )), + ); + let (tracker_deleter, tracker_deleter_rx) = DeadlinerTask::start( + ct.clone(), + "tracker_deleter", + Arc::new(OffsetCalculator::new( + Arc::clone(&deadline_calc), + tracker_lag.saturating_add(std::time::Duration::from_secs(60)), + )), + ); + + let tracker_feature_set = tracker_feature_set(&feature_set); + + let track_from = calculate_tracker_delay(ð2_cl, slot_duration).await?; + let tracker = TrackerService::start( + ct.clone(), + tracker_analyser, + AnalyserRx(tracker_analyser_rx), + tracker_deleter, + DeleterRx(tracker_deleter_rx), + peers, + track_from, + Arc::clone(&tracker_feature_set), + ); + + // Resolves the terminal `ChainInclusion` step; without it every duty with an + // inclusion step would stall unresolved and be reported as failed. Spawned + // and supervised by `run_lifecycle`. + let inclusion_checker = { + let tracker = Arc::clone(&tracker); + Arc::new( + InclusionChecker::new( + eth2_cl.clone(), + Box::new(move |duty: &Duty, pubkey: PubKey, err| { + let tracker = Arc::clone(&tracker); + let duty = duty.clone(); + // The core's callback is sync but `inclusion_checked` is + // async, so hand the event to the runtime. + tokio::spawn(async move { + tracker.inclusion_checked(duty, pubkey, err).await; + }); + }), + Arc::clone(&tracker_feature_set), + ) + .await + .map_err(AppError::BeaconApi)?, + ) + }; + // ---- (4) AggSigDB (built before fetcher: agg_sig_db back-edge target) ---- let aggsigdb = MemoryDBHandle::new(aggsigdb_deadliner, aggsigdb_deadliner_rx, ct.clone()); @@ -364,22 +587,38 @@ pub async fn wire_core_workflow( let consensus = Arc::clone(&consensus); let ct = ct.clone(); let deadline_calc = Arc::clone(&deadline_calc); + let tracker = Arc::clone(&tracker); Arc::new(move |duty: Duty, set: UnsignedDataSet| { let consensus = Arc::clone(&consensus); let ct = ct.clone(); let deadline_calc = Arc::clone(&deadline_calc); + let tracker = Arc::clone(&tracker); Box::pin(async move { + let pubkeys: Vec = set.keys().copied().collect(); let value = unsigneddata::unsigned_data_set_to_proto(&set)?; // Bound consensus by the duty deadline so a stuck instance is // cancelled (-> ConsensusTimeout) instead of running until shutdown. - run_bounded_by_duty_deadline( + let result = run_bounded_by_duty_deadline( &deadline_calc, &ct, - duty, + duty.clone(), move |duty, dct| async move { consensus.propose(duty, value, &dct).await }, ) - .await?; - Ok(()) + .await; + + match result { + Ok(()) => { + tracker.consensus_proposed(duty, &pubkeys, None).await; + Ok(()) + } + Err(err) => { + let (reported, returned) = share_step_err(err); + tracker + .consensus_proposed(duty, &pubkeys, Some(reported)) + .await; + Err(returned.into()) + } + } }) }) }; @@ -404,8 +643,10 @@ pub async fn wire_core_workflow( // we spawn the async `dutydb.store` inside it. { let dutydb = Arc::clone(&dutydb); + let tracker = Arc::clone(&tracker); consensus.subscribe(move |duty: Duty, value: pbcore::UnsignedDataSet| { let dutydb = Arc::clone(&dutydb); + let tracker = Arc::clone(&tracker); tokio::spawn(async move { let core_set = match unsigneddata::unsigned_data_set_from_proto(&duty.duty_type, &value) { @@ -415,9 +656,16 @@ pub async fn wire_core_workflow( return; } }; - if let Err(err) = dutydb.store(duty, core_set).await { - tracing::warn!(?err, "dutydb: store"); - } + let pubkeys: Vec = core_set.keys().copied().collect(); + // Logged before the error moves into the tracker's `Arc`. + let step_err = match dutydb.store(duty.clone(), core_set).await { + Ok(()) => None, + Err(err) => { + tracing::warn!(?err, "dutydb: store"); + Some(owned_step_err(err)) + } + }; + tracker.duty_db_stored(duty, &pubkeys, step_err).await; }); Ok(()) }); @@ -433,17 +681,31 @@ pub async fn wire_core_workflow( // Stitch: parsigdb.subscribe_internal(parsigex.broadcast). { let broadcast = Arc::clone(&parsigex.broadcast); + let tracker = Arc::clone(&tracker); parsigdb .subscribe_internal(parsigdb::memory::internal_subscriber( move |duty: Duty, set: ParSignedDataSet| { let broadcast = Arc::clone(&broadcast); + let tracker = Arc::clone(&tracker); async move { - broadcast(duty, set).await.map_err(|e| { - parsigdb::memory::InternalSubscriberError::ParsigexBroadcast { - source: Box::new(e), + match broadcast(duty.clone(), set.clone()).await { + Ok(()) => { + tracker.par_sig_ex_broadcasted(duty, &set, None).await; + Ok(()) } - .into() - }) + Err(err) => { + let (reported, returned) = share_step_err(err); + tracker + .par_sig_ex_broadcasted(duty, &set, Some(reported)) + .await; + Err( + parsigdb::memory::InternalSubscriberError::ParsigexBroadcast { + source: Box::new(returned), + } + .into(), + ) + } + } } }, )) @@ -462,14 +724,22 @@ pub async fn wire_core_workflow( // sinks; in Charon they are wrapped in async-retry subscribers — part B). { let aggsigdb = aggsigdb.clone(); + let tracker = Arc::clone(&tracker); aggregator.subscribe(Arc::new(move |duty: &Duty, set: &SignedDataSet| { let aggsigdb = aggsigdb.clone(); + let tracker = Arc::clone(&tracker); let duty = duty.clone(); let set = set.clone(); Box::pin(async move { - if let Err(err) = aggsigdb.store(duty, set).await { - tracing::warn!(?err, "aggsigdb: store"); - } + let pubkeys: Vec = set.keys().copied().collect(); + let step_err = match aggsigdb.store(duty.clone(), set).await { + Ok(()) => None, + Err(err) => { + tracing::warn!(?err, "aggsigdb: store"); + Some(owned_step_err(err)) + } + }; + tracker.agg_sig_db_stored(duty, &pubkeys, step_err).await; Ok(()) }) })); @@ -483,14 +753,40 @@ pub async fn wire_core_workflow( // Stitch: sigagg.subscribe(broadcaster.broadcast). { let broadcaster = Arc::clone(&broadcaster); + let tracker = Arc::clone(&tracker); + let inclusion = Arc::clone(&inclusion_checker); aggregator.subscribe(Arc::new(move |duty: &Duty, set: &SignedDataSet| { let broadcaster = Arc::clone(&broadcaster); + let tracker = Arc::clone(&tracker); + let inclusion = Arc::clone(&inclusion); let duty = duty.clone(); let set = set.clone(); Box::pin(async move { - if let Err(err) = broadcaster.broadcast(duty, set).await { - tracing::warn!(?err, "broadcaster: broadcast"); + let pubkeys: Vec = set.keys().copied().collect(); + + // Register for inclusion checking before broadcasting, and even + // if the broadcast fails: peers may still succeed, so the duty + // can land on-chain regardless. Parity: charon + // `core/tracking.go` `BroadcasterBroadcast`. + if let Err(err) = inclusion.submitted(&duty, &set) { + tracing::error!( + ?err, + duty = %duty, + "Internal error: failed to submit duty to inclusion checker. \ + This indicates a tracking bug that should be reported", + ); } + + let step_err = match broadcaster.broadcast(duty.clone(), set).await { + Ok(()) => None, + Err(err) => { + tracing::warn!(?err, "broadcaster: broadcast"); + Some(owned_step_err(err)) + } + }; + tracker + .broadcaster_broadcast(duty, &pubkeys, step_err) + .await; Ok(()) }) })); @@ -504,28 +800,47 @@ pub async fn wire_core_workflow( // spawning the aggregation and awaiting its `JoinHandle` (which is `Sync`). { let aggregator = Arc::clone(&aggregator); + let tracker_agg = Arc::clone(&tracker); parsigdb .subscribe_threshold(parsigdb::memory::threshold_subscriber( move |duty: Duty, set: HashMap>| { let aggregator = Arc::clone(&aggregator); + let tracker = Arc::clone(&tracker_agg); async move { + let pubkeys: Vec = set.keys().copied().collect(); + let tracked = duty.clone(); let result = tokio::spawn(async move { aggregator.aggregate(&duty, &set).await }) .await; match result { - Ok(Ok(())) => Ok(()), - Ok(Err(e)) => Err( - parsigdb::memory::InternalSubscriberError::ParsigexBroadcast { - source: Box::new(e), - } - .into(), - ), - Err(e) => Err( - parsigdb::memory::InternalSubscriberError::ParsigexBroadcast { - source: Box::new(e), - } - .into(), - ), + Ok(Ok(())) => { + tracker.sig_agg_aggregated(tracked, &pubkeys, None).await; + Ok(()) + } + Ok(Err(e)) => { + let (reported, returned) = share_step_err(e); + tracker + .sig_agg_aggregated(tracked, &pubkeys, Some(reported)) + .await; + Err( + parsigdb::memory::InternalSubscriberError::ParsigexBroadcast { + source: Box::new(returned), + } + .into(), + ) + } + Err(e) => { + let (reported, returned) = share_step_err(e); + tracker + .sig_agg_aggregated(tracked, &pubkeys, Some(reported)) + .await; + Err( + parsigdb::memory::InternalSubscriberError::ParsigexBroadcast { + source: Box::new(returned), + } + .into(), + ) + } } } }, @@ -536,12 +851,21 @@ pub async fn wire_core_workflow( // ---- (9) parsigex inbound subscription -> parsigdb.store_external ---- { let parsigdb = Arc::clone(&parsigdb); + let tracker_ext = Arc::clone(&tracker); let received: ParSigExReceived = Arc::new(move |duty: Duty, set: ParSignedDataSet| { let parsigdb = Arc::clone(&parsigdb); + let tracker = Arc::clone(&tracker_ext); Box::pin(async move { - if let Err(err) = parsigdb.store_external(&duty, &set).await { - tracing::warn!(?err, "parsigdb: store external"); - } + let step_err = match parsigdb.store_external(&duty, &set).await { + Ok(()) => None, + Err(err) => { + tracing::warn!(?err, "parsigdb: store external"); + Some(owned_step_err(err)) + } + }; + tracker + .par_sig_db_stored_external(duty, &set, step_err) + .await; }) }); (parsigex.subscribe)(received).await; @@ -558,14 +882,17 @@ pub async fn wire_core_workflow( { let fetcher = Arc::clone(&fetcher); let ct = ct.clone(); + let tracker = Arc::clone(&tracker); sched_builder.subscribe_duty( move |duty: &Duty, set: &pluto_core::types::DutyDefinitionSet| { let fetcher = Arc::clone(&fetcher); let ct = ct.clone(); + let tracker = Arc::clone(&tracker); let duty = duty.clone(); let set = set.clone(); async move { - match fetcher.fetch(duty, set).await { + let pubkeys: Vec = set.keys().copied().collect(); + match fetcher.fetch(duty.clone(), set).await { // In-flight fetches racing shutdown fail against already // terminated components (e.g. the aggsigdb back-edge); // don't surface those as duty errors. @@ -573,7 +900,19 @@ pub async fn wire_core_workflow( tracing::debug!(?err, "fetch aborted by shutdown"); Ok(()) } - res => res, + Ok(()) => { + tracker.fetcher_fetched(duty, &pubkeys, None).await; + Ok(()) + } + Err(err) => { + let (reported, returned) = share_step_err(err); + tracker + .fetcher_fetched(duty, &pubkeys, Some(reported)) + .await; + // `subscribe_duty` is generic over the error type, so + // the shared wrapper propagates as-is. + Err(returned) + } } } }, @@ -711,13 +1050,24 @@ pub async fn wire_core_workflow( // Stitch: vapi.subscribe(parsigdb.store_internal). { let parsigdb = Arc::clone(&parsigdb); + let tracker = Arc::clone(&tracker); vapi.subscribe(move |duty: Duty, set: ParSignedDataSet| { let parsigdb = Arc::clone(&parsigdb); + let tracker = Arc::clone(&tracker); async move { - parsigdb - .store_internal(&duty, &set) - .await - .map_err(Into::into) + match parsigdb.store_internal(&duty, &set).await { + Ok(()) => { + tracker.par_sig_db_stored_internal(duty, &set, None).await; + Ok(()) + } + Err(err) => { + let (reported, returned) = share_step_err(err); + tracker + .par_sig_db_stored_internal(duty, &set, Some(reported)) + .await; + Err(returned.into()) + } + } } }); } @@ -740,6 +1090,7 @@ pub async fn wire_core_workflow( parsigdb_deadliner_rx, aggsigdb, fetcher, + inclusion_checker, validator_api_router, }) } @@ -1176,4 +1527,34 @@ mod tests { ); assert!(active.contains_key(&1)); } + + fn feature_set(enabled: Vec) -> Arc { + Arc::new( + FeatureSet::from_config(pluto_featureset::Config { + enabled, + ..Default::default() + }) + .expect("valid featureset"), + ) + } + + /// `AttestationInclusion` is masked off for the tracker so the + /// proposer-only inclusion checker is never fed attester/aggregator + /// submissions. + #[test] + fn tracker_feature_set_masks_attestation_inclusion() { + let fs = feature_set(vec![Feature::AttestationInclusion]); + assert!(fs.enabled(Feature::AttestationInclusion)); + + let tracker_fs = tracker_feature_set(&fs); + assert!(!tracker_fs.enabled(Feature::AttestationInclusion)); + } + + /// Without the feature the set is passed through untouched (same `Arc`). + #[test] + fn tracker_feature_set_is_passthrough_when_disabled() { + let fs = feature_set(vec![]); + let tracker_fs = tracker_feature_set(&fs); + assert!(Arc::ptr_eq(&fs, &tracker_fs)); + } } diff --git a/crates/app/tests/wiring.rs b/crates/app/tests/wiring.rs index 9580aa8b..c3c9e2c4 100644 --- a/crates/app/tests/wiring.rs +++ b/crates/app/tests/wiring.rs @@ -269,6 +269,13 @@ fn wire_inputs_with( fetch_only_comm_idx0: false, seen_pubkeys: None, slot_tick: None, + // Single-node wiring test: one peer at share index 1, default features + // (so only proposers carry an on-chain inclusion step). + peers: vec![pluto_core::tracker::PeerInfo { + name: "test".to_string(), + share_idx: 1, + }], + feature_set: Arc::new(pluto_featureset::FeatureSet::default()), } } diff --git a/crates/core/src/tracker/inclusion.rs b/crates/core/src/tracker/inclusion.rs index 88b7d605..a7947e61 100644 --- a/crates/core/src/tracker/inclusion.rs +++ b/crates/core/src/tracker/inclusion.rs @@ -15,11 +15,21 @@ // reporters, committee plumbing) have no in-crate caller. #![allow(dead_code)] -use std::{any::Any, collections::HashMap, sync::Arc, time::Duration}; +use std::{ + any::Any, + collections::HashMap, + sync::{Arc, Mutex}, + time::Duration, +}; -use pluto_eth2api::versioned; +use chrono::{DateTime, Utc}; +use pluto_eth2api::{ + EthBeaconNodeApiClient, EthBeaconNodeApiClientError, GetBlockV2Request, GetBlockV2Response, + versioned, +}; use pluto_featureset::FeatureSet; use pluto_ssz::{BitList, HashRoot}; +use tokio_util::sync::CancellationToken; use tree_hash::TreeHash; use crate::{ @@ -28,12 +38,23 @@ use crate::{ VersionedSignedAggregateAndProof, VersionedSignedProposal, }, tracker::{StepError, analysis::incl_supported, metrics::TRACKER_METRICS}, - types::{Duty, DutyType, PubKey, SignedData}, + types::{Duty, DutyType, PubKey, SignedData, SignedDataSet}, }; /// Number of slots after which an unincluded duty is assumed missed and its /// cached submission (and associated committee state) is dropped. -const INCL_MISSED_LAG: u64 = 32; +/// +/// Public because the tracker's analyser deadline is offset by +/// `INCL_MISSED_LAG + INCL_CHECK_LAG` slots so that a duty is analysed only +/// after its inclusion verdict can have arrived. Parity: charon +/// `core/tracker/inclusion.go` `InclMissedLag`, consumed by `app.go`. +pub const INCL_MISSED_LAG: u64 = 32; + +/// Number of slots to lag behind the head before checking inclusion, giving the +/// beacon node time to import the block for that slot. +/// +/// Parity: charon `core/tracker/inclusion.go` `InclCheckLag`. +pub const INCL_CHECK_LAG: u64 = 6; /// SSZ capacity bound used only to decode aggregation bitlists for bit-level /// comparisons. The bit operations (`contains`/`bit_at`) work on the decoded @@ -301,14 +322,8 @@ impl InclusionCore { .iter() .filter_map(|(key, sub)| match sub.duty.duty_type { DutyType::Proposer => (sub.duty.slot.inner() == slot).then(|| key.clone()), - // Parity: charon core/tracker/inclusion.go:289-291 @ v1.7.1 - // panics with "bug: unexpected type" here — CheckBlock (the - // non-attestation path) is only ever fed proposer submissions. - // `unreachable!` reproduces that panic with the same message. - // Accepted divergence in panic *site* only: Go panics while - // iterating the offending submission; Rust panics inside the - // `filter_map` closure on the same element — observably - // identical. + // CheckBlock is only ever fed proposer submissions. Parity: + // charon core/tracker/inclusion.go panics identically here. _ => unreachable!("bug: unexpected type"), }) .collect(); @@ -602,6 +617,167 @@ fn log_block_included(sub: &Submission, block_slot: u64, blinded: bool) { ); } +/// Networked driver that polls the beacon node and feeds [`InclusionCore`]. +/// +/// Parity: charon's `InclusionChecker` in `core/tracker/inclusion.go`. The +/// core holds all the decision logic; this type owns only the clock, the +/// beacon-node calls, and the once-per-slot cadence. +pub struct InclusionChecker { + core: Mutex, + eth2_cl: EthBeaconNodeApiClient, + genesis: DateTime, + slot_duration: Duration, +} + +impl InclusionChecker { + /// Builds a checker, fetching genesis time and slot duration up front (as + /// charon's `NewInclusion` does) so the ticker loop needs no further config + /// lookups. + pub async fn new( + eth2_cl: EthBeaconNodeApiClient, + tracker_incl_fn: TrackerInclFn, + feature_set: Arc, + ) -> Result { + let genesis = eth2_cl.fetch_genesis_time().await?; + let (slot_duration, _slots_per_epoch) = eth2_cl.fetch_slots_config().await?; + + Ok(Self { + core: Mutex::new(InclusionCore::new(tracker_incl_fn, feature_set)), + eth2_cl, + genesis, + slot_duration, + }) + } + + /// Records a duty submitted to the beacon node, stamping each entry with + /// the delay between slot start and broadcast. + /// + /// Synchronous (the core is I/O-free) so the broadcaster stitch point can + /// call it without awaiting. + pub fn submitted(&self, duty: &Duty, set: &SignedDataSet) -> Result<(), InclusionError> { + let delay = self.delay_since_slot_start(duty.slot.inner()); + + let mut core = self.core.lock().expect("inclusion core mutex poisoned"); + for (pubkey, data) in set { + core.submitted(duty.clone(), *pubkey, data.clone(), delay)?; + } + + Ok(()) + } + + /// Elapsed time between the start of `slot` and now, saturating at zero for + /// a slot that has not started yet. + fn delay_since_slot_start(&self, slot: u64) -> Duration { + let offset = self + .slot_duration + .saturating_mul(u32::try_from(slot).unwrap_or(u32::MAX)); + let Some(slot_start) = chrono::Duration::from_std(offset) + .ok() + .and_then(|offset| self.genesis.checked_add_signed(offset)) + else { + return Duration::ZERO; + }; + Utc::now() + .signed_duration_since(slot_start) + .to_std() + .unwrap_or(Duration::ZERO) + } + + /// Slot currently due for an inclusion check: the head slot lagged by + /// [`INCL_CHECK_LAG`] so the beacon node has had time to import it. `None` + /// before the chain is old enough to have such a slot. + fn check_due_slot(&self) -> Option { + let elapsed = Utc::now() + .signed_duration_since(self.genesis) + .to_std() + .ok()?; + let head = u64::try_from( + elapsed + .as_nanos() + .checked_div(self.slot_duration.as_nanos())?, + ) + .ok()?; + head.checked_sub(INCL_CHECK_LAG) + } + + /// Reports whether a block exists at `slot`. A `404` means no block was + /// proposed, which is a normal outcome rather than an error — the same + /// distinction charon draws via `is404Error`. + async fn block_exists(&self, slot: u64) -> Result { + let request = GetBlockV2Request::builder() + .block_id(slot.to_string()) + .build() + .map_err(|err| InclusionCheckerError::Request(err.into()))?; + + match self + .eth2_cl + .get_block_v2(request) + .await + .map_err(|err| InclusionCheckerError::Request(err.into()))? + { + GetBlockV2Response::Ok(_) | GetBlockV2Response::OkBinary(_) => Ok(true), + GetBlockV2Response::NotFound(_) => Ok(false), + other => Err(InclusionCheckerError::UnexpectedResponse(format!( + "{other:?}" + ))), + } + } + + /// Drives inclusion checking until `cancel` fires: once per due slot, ask + /// the beacon node whether that slot produced a block, feed the verdict to + /// the core, then trim submissions old enough to count as missed. + pub async fn run(self: Arc, cancel: CancellationToken) { + let mut ticker = tokio::time::interval(Duration::from_secs(1)); + let mut checked_slot: Option = None; + + loop { + tokio::select! { + () = cancel.cancelled() => return, + _ = ticker.tick() => {} + } + + let Some(slot) = self.check_due_slot() else { + continue; + }; + if checked_slot == Some(slot) { + continue; + } + + let found = match self.block_exists(slot).await { + Ok(found) => found, + Err(err) => { + tracing::warn!(slot, %err, "Failed to check inclusion"); + continue; + } + }; + + { + let mut core = self.core.lock().expect("inclusion core mutex poisoned"); + core.check_block(slot, found); + if let Some(trim_to) = slot.checked_sub(INCL_MISSED_LAG) { + core.trim(trim_to); + } + } + + checked_slot = Some(slot); + } + } +} + +/// Errors raised by the networked [`InclusionChecker`] while talking to the +/// beacon node. Logged and skipped rather than propagated: the slot is just +/// retried on the next tick. +#[derive(Debug, thiserror::Error)] +pub enum InclusionCheckerError { + /// The beacon-node request failed or could not be built. Boxed because the + /// generated client surfaces `anyhow::Error`, which `pluto-core` avoids. + #[error("beacon node request failed: {0}")] + Request(#[source] Box), + /// The beacon node returned a status the checker does not handle. + #[error("unexpected beacon node response: {0}")] + UnexpectedResponse(String), +} + #[cfg(test)] mod tests { use std::{ diff --git a/crates/core/src/validatorapi/component.rs b/crates/core/src/validatorapi/component.rs index 3da601ef..0e9bd91c 100644 --- a/crates/core/src/validatorapi/component.rs +++ b/crates/core/src/validatorapi/component.rs @@ -529,7 +529,7 @@ impl Component { self.dutydb .await_proposal(slot) .await - .map_err(map_dutydb_error) + .map_err(|err| map_dutydb_error("proposal", err)) } /// Resolves the validator index for a VC-submitted attestation. @@ -1067,7 +1067,7 @@ impl Handler for Component { "attestation data not available before deadline", ) })? - .map_err(map_dutydb_error)?; + .map_err(|err| map_dutydb_error("attestation", err))?; Ok(AttestationDataResponse { data }) } @@ -2045,20 +2045,27 @@ fn upstream_unexpected(endpoint: &'static str, response: R) } /// Maps a [`crate::dutydb::Error`] into the `ApiError` returned to the client -/// when an `attestation_data` await fails. `Shutdown` propagates as 503 so the -/// VC can retry; `AwaitDutyExpired` propagates as 408 — same as a timeout — -/// since the duty is gone and the data will never arrive. Anything else is a -/// programming error here and becomes 500. -fn map_dutydb_error(err: DutyDbError) -> ApiError { +/// when a duty-data await fails. `Shutdown` propagates as 503 so the VC can +/// retry; `AwaitDutyExpired` propagates as 408 — same as a timeout — since the +/// duty is gone and the data will never arrive. Anything else is a programming +/// error here and becomes 500. +/// +/// `duty` names the duty being awaited (e.g. `"attestation"`, `"proposal"`) so +/// the client-visible message matches the request that produced it; this mapper +/// is shared by more than one endpoint. +fn map_dutydb_error(duty: &'static str, err: DutyDbError) -> ApiError { let (status, message) = match err { - DutyDbError::Shutdown => (StatusCode::SERVICE_UNAVAILABLE, "dutydb is shutting down"), + DutyDbError::Shutdown => ( + StatusCode::SERVICE_UNAVAILABLE, + "dutydb is shutting down".to_string(), + ), DutyDbError::AwaitDutyExpired => ( StatusCode::REQUEST_TIMEOUT, - "attestation duty expired before data was stored", + format!("{duty} duty expired before data was stored"), ), _ => ( StatusCode::INTERNAL_SERVER_ERROR, - "await attestation failed", + format!("await {duty} failed"), ), }; ApiError::new(status, message).with_source(err) @@ -3169,19 +3176,35 @@ mod tests { #[test] fn map_dutydb_error_status_codes() { assert_eq!( - map_dutydb_error(DutyDbError::Shutdown).status_code, + map_dutydb_error("attestation", DutyDbError::Shutdown).status_code, StatusCode::SERVICE_UNAVAILABLE ); assert_eq!( - map_dutydb_error(DutyDbError::AwaitDutyExpired).status_code, + map_dutydb_error("attestation", DutyDbError::AwaitDutyExpired).status_code, StatusCode::REQUEST_TIMEOUT ); assert_eq!( - map_dutydb_error(DutyDbError::UnsupportedDutyType).status_code, + map_dutydb_error("attestation", DutyDbError::UnsupportedDutyType).status_code, StatusCode::INTERNAL_SERVER_ERROR ); } + /// The expiry message names the duty that was awaited: the mapper is shared + /// by `attestation_data` and `await_proposal`, and previously reported + /// every timeout — including a missed block proposal — as an + /// attestation. + #[test] + fn map_dutydb_error_message_names_the_duty() { + assert_eq!( + map_dutydb_error("proposal", DutyDbError::AwaitDutyExpired).message, + "proposal duty expired before data was stored" + ); + assert_eq!( + map_dutydb_error("attestation", DutyDbError::AwaitDutyExpired).message, + "attestation duty expired before data was stored" + ); + } + /// `upstream_status_error` keeps the upstream response body out of the /// client-visible message but preserves it on `source()` so it lands in /// the debug log. diff --git a/crates/core/src/validatorapi/error.rs b/crates/core/src/validatorapi/error.rs index e13c440d..f119c49b 100644 --- a/crates/core/src/validatorapi/error.rs +++ b/crates/core/src/validatorapi/error.rs @@ -100,6 +100,28 @@ struct ErrorBody { impl IntoResponse for ApiError { fn into_response(self) -> Response { + // The `source` never reaches the client (it can carry internal detail), + // but it is the only place the underlying cause is recorded — e.g. which + // field an SSZ/JSON body failed to decode. Log it here, on the single + // path every error response takes, otherwise it is silently dropped. + if let Some(source) = &self.source { + if self.status_code.is_server_error() { + tracing::warn!( + status = self.status_code.as_u16(), + message = %self.message, + source = %DisplayChain(source.as_ref()), + "validator api error" + ); + } else { + tracing::debug!( + status = self.status_code.as_u16(), + message = %self.message, + source = %DisplayChain(source.as_ref()), + "validator api error" + ); + } + } + let body = ErrorBody { code: self.status_code.as_u16(), message: self.message, @@ -108,3 +130,20 @@ impl IntoResponse for ApiError { (self.status_code, Json(body)).into_response() } } + +/// Renders an error together with its `source()` chain, so a wrapped cause +/// (such as the inner `ssz::DecodeError` behind a decode failure) is not +/// truncated to just the outermost message. +struct DisplayChain<'a>(&'a (dyn std::error::Error + 'static)); + +impl fmt::Display for DisplayChain<'_> { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.0)?; + let mut current = self.0.source(); + while let Some(err) = current { + write!(f, ": {err}")?; + current = err.source(); + } + Ok(()) + } +} diff --git a/crates/eth2api/src/spec/bellatrix.rs b/crates/eth2api/src/spec/bellatrix.rs index 4162d27e..ed8e552d 100644 --- a/crates/eth2api/src/spec/bellatrix.rs +++ b/crates/eth2api/src/spec/bellatrix.rs @@ -22,9 +22,22 @@ pub const MAX_BYTES_PER_TRANSACTION: usize = 1_073_741_824; pub type BaseFeePerGas = U256; /// Raw execution transaction bytes. +/// +/// Spec: `Transaction = ByteList[MAX_BYTES_PER_TRANSACTION]` — a bare SSZ list, +/// not a container. `struct_behaviour = "transparent"` is therefore required: +/// without it `ssz_derive` frames every transaction as a single-field container +/// (a spurious 4-byte offset prefix), which makes any block carrying at least +/// one transaction fail to decode. Charon models the same type as a plain +/// `type Transaction []byte`. +/// +/// `TreeHash` intentionally keeps the derived container behaviour: merkleizing +/// a one-field container yields the single leaf unchanged, so the root already +/// equals the bare list's root (locked in by the `*_beacon_block_body_root` +/// spec-vector tests, whose fixtures carry transactions). #[serde_as] #[derive(Debug, Clone, PartialEq, Eq, Encode, Decode, TreeHash, Serialize, Deserialize)] #[serde(transparent)] +#[ssz(struct_behaviour = "transparent")] pub struct Transaction { /// Transaction bytes. #[serde_as(as = "pluto_ssz::serde_utils::Hex0x")] @@ -358,4 +371,53 @@ mod tests { fn json_matches_vector(actual: serde_json::Value, expected_json: &'static str) { test_fixtures::assert_json_eq(actual, expected_json); } + + /// `Transaction` is `ByteList[MAX_BYTES_PER_TRANSACTION]`, so its SSZ form + /// is the raw transaction bytes with no framing. Without + /// `struct_behaviour = "transparent"` the derive emitted a leading 4-byte + /// offset, which made every block carrying a transaction undecodable. + #[test] + fn transaction_ssz_is_the_bare_byte_list() { + use ssz::{Decode, Encode}; + + // A realistic type-3 (blob) transaction prefix: the first four bytes are + // not a valid container offset, which is exactly why container framing + // broke decoding. + let raw = vec![0x03, 0xf8, 0xb9, 0x83, 0xde, 0xad, 0xbe, 0xef]; + let tx = super::Transaction::from(raw.clone()); + + assert_eq!(tx.as_ssz_bytes(), raw, "encoding must not add framing"); + assert_eq!(tx.ssz_bytes_len(), raw.len()); + assert_eq!( + super::Transaction::from_ssz_bytes(&raw).expect("raw bytes decode"), + tx + ); + // Empty transactions are a valid zero-length list. + assert!(super::Transaction::from_ssz_bytes(&[]).is_ok()); + } + + /// An execution payload carrying transactions must survive an SSZ round + /// trip. Previously the fixtures only exercised `TreeHash` and JSON, so + /// the broken SSZ framing went unnoticed. + #[test] + fn execution_payload_with_transactions_ssz_round_trips() { + use ssz::{Decode, Encode}; + + let payload = test_fixtures::bellatrix_execution_payload_fixture(); + assert!( + !payload.transactions.0.is_empty(), + "fixture must carry transactions for this to be meaningful" + ); + + let bytes = payload.as_ssz_bytes(); + let decoded = + super::ExecutionPayload::from_ssz_bytes(&bytes).expect("payload round trips via ssz"); + assert_eq!(decoded, payload); + + // The transactions list is `List[Transaction, N]`: a 4-byte offset per + // element followed by each element's bare bytes. + let txs = &payload.transactions.0; + let expected_len = txs.len() * 4 + txs.iter().map(|tx| tx.bytes.0.len()).sum::(); + assert_eq!(payload.transactions.as_ssz_bytes().len(), expected_len); + } }