From c995f2b39a90d1c552a1efb80f3696b6dbc4c40f Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Thu, 20 Aug 2026 16:02:04 -0400 Subject: [PATCH 01/12] wip --- proto/drand_pb.proto | 8 ++ src/beacon/drand_pb.rs | 54 +++++++++++++ src/beacon/mod.rs | 2 + src/chain_sync/chain_follower.rs | 12 +++ src/chain_sync/metrics.rs | 2 + src/libp2p/behaviour.rs | 19 ++--- src/libp2p/gossip_params.rs | 13 ++-- src/libp2p/service.rs | 126 +++++++++++++++++++++++-------- src/networks/mod.rs | 18 ++++- 9 files changed, 204 insertions(+), 50 deletions(-) create mode 100644 proto/drand_pb.proto create mode 100644 src/beacon/drand_pb.rs diff --git a/proto/drand_pb.proto b/proto/drand_pb.proto new file mode 100644 index 000000000000..db63e3120baa --- /dev/null +++ b/proto/drand_pb.proto @@ -0,0 +1,8 @@ +syntax = "proto3"; + +package drand_pb; + +message PublicRandResponse { + uint64 round = 1; + bytes signature = 2; +} diff --git a/src/beacon/drand_pb.rs b/src/beacon/drand_pb.rs new file mode 100644 index 000000000000..b36624fcb421 --- /dev/null +++ b/src/beacon/drand_pb.rs @@ -0,0 +1,54 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT +// Automatically generated rust module for 'drand_pb.proto' file +// Command: `pb-rs -s -D proto/drand_pb.proto`, See + +#![allow(non_snake_case)] +#![allow(non_upper_case_globals)] +#![allow(non_camel_case_types)] +#![allow(unused_imports)] +#![allow(unknown_lints)] +#![allow(clippy::all)] +#![cfg_attr(rustfmt, rustfmt_skip)] + + +use quick_protobuf::{MessageInfo, MessageRead, MessageWrite, BytesReader, Writer, WriterBackend, Result}; +use quick_protobuf::sizeofs::*; +use super::*; + +#[allow(clippy::derive_partial_eq_without_eq)] +#[derive(Debug, Default, PartialEq, Clone)] +pub struct PublicRandResponse { + pub round: u64, + pub signature: Vec, +} + +impl<'a> MessageRead<'a> for PublicRandResponse { + fn from_reader(r: &mut BytesReader, bytes: &'a [u8]) -> Result { + let mut msg = Self::default(); + while !r.is_eof() { + match r.next_tag(bytes) { + Ok(8) => msg.round = r.read_uint64(bytes)?, + Ok(18) => msg.signature = r.read_bytes(bytes)?.to_owned(), + Ok(t) => { r.read_unknown(bytes, t)?; } + Err(e) => return Err(e), + } + } + Ok(msg) + } +} + +impl MessageWrite for PublicRandResponse { + fn get_size(&self) -> usize { + 0 + + if self.round == 0u64 { 0 } else { 1 + sizeof_varint(*(&self.round) as u64) } + + if self.signature.is_empty() { 0 } else { 1 + sizeof_len((&self.signature).len()) } + } + + fn write_message(&self, w: &mut Writer) -> Result<()> { + if self.round != 0u64 { w.write_with_tag(8, |w| w.write_uint64(*&self.round))?; } + if !self.signature.is_empty() { w.write_with_tag(18, |w| w.write_bytes(&**&self.signature))?; } + Ok(()) + } +} + diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index 8718f77407fa..b4b536c4f5d4 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -3,9 +3,11 @@ pub mod beacon_entries; mod drand; +mod drand_pb; pub mod signatures; pub use beacon_entries::*; pub use drand::*; +pub use drand_pb::PublicRandResponse; #[cfg(test)] pub mod mock_beacon; diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index bc57fd569d2c..d79c89d91de2 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -304,6 +304,17 @@ async fn chain_follower( debug!("Received invalid GossipSub message: {}", why); } } + PubsubMessage::DrandEntry { + chain_hash, + response, + } => { + // Signature verification and cache insertion land with the + // beacon-schedule wiring. + tracing::trace!( + "Received drand round {} on chain {chain_hash}", + response.round + ); + } }, _ => {} } @@ -480,6 +491,7 @@ fn inc_gossipsub_event_metrics(event: &NetworkEvent) { NetworkEvent::PubsubMessage { message } => match message { PubsubMessage::Block(_) => metrics::values::PUBSUB_BLOCK, PubsubMessage::Message(_) => metrics::values::PUBSUB_MESSAGE, + PubsubMessage::DrandEntry { .. } => metrics::values::PUBSUB_DRAND_ENTRY, }, NetworkEvent::ChainExchangeRequestOutbound => { metrics::values::CHAIN_EXCHANGE_REQUEST_OUTBOUND diff --git a/src/chain_sync/metrics.rs b/src/chain_sync/metrics.rs index 4f6c715aa7f3..78fdc4431de2 100644 --- a/src/chain_sync/metrics.rs +++ b/src/chain_sync/metrics.rs @@ -94,6 +94,8 @@ pub mod values { Libp2pMessageKindLabel::new("pubsub_message_block"); pub const PUBSUB_MESSAGE: Libp2pMessageKindLabel = Libp2pMessageKindLabel::new("pubsub_message_message"); + pub const PUBSUB_DRAND_ENTRY: Libp2pMessageKindLabel = + Libp2pMessageKindLabel::new("pubsub_message_drand_entry"); pub const CHAIN_EXCHANGE_REQUEST_OUTBOUND: Libp2pMessageKindLabel = Libp2pMessageKindLabel::new("chain_exchange_request_out"); pub const CHAIN_EXCHANGE_RESPONSE_INBOUND: Libp2pMessageKindLabel = diff --git a/src/libp2p/behaviour.rs b/src/libp2p/behaviour.rs index df9c1c1cda31..3a2b36aba095 100644 --- a/src/libp2p/behaviour.rs +++ b/src/libp2p/behaviour.rs @@ -10,7 +10,7 @@ use super::{ PeerManager, discovery::{DerivedDiscoveryBehaviourEvent, DiscoveryEvent, PeerInfo}, }; -use crate::libp2p_bitswap::BitswapBehaviour; +use crate::{libp2p::PubsubTopicCfg, libp2p_bitswap::BitswapBehaviour}; use crate::utils::{encoding::blake2b_256, version::FOREST_VERSION_STRING}; use crate::{ libp2p::{ @@ -79,11 +79,13 @@ const MAX_SUBSCRIPTIONS_PER_REQUEST: usize = 100; /// Filter accepting only Forest's topics, bounded in count and per request. pub(in crate::libp2p) fn build_subscription_filter( - network_name: &GenesisNetworkName, + cfg: PubsubTopicCfg<'_> ) -> MaxCountSubscriptionFilter { - let allowed: Vec<_> = crate::libp2p::pubsub_topics(network_name) - .map(|t| t.hash()) + let allowed: Vec<_> = crate::libp2p::pubsub_topics(cfg) + .iter() + .map(|(_, t)| t.hash()) .collect(); + MaxCountSubscriptionFilter { // Whitelisted topics are the only ones counted, so their number is an // exact, self-maintaining bound. @@ -95,7 +97,7 @@ pub(in crate::libp2p) fn build_subscription_filter( pub(in crate::libp2p) fn build_gossipsub( local_key: &Keypair, - network_name: &GenesisNetworkName, + cfg: PubsubTopicCfg<'_> ) -> anyhow::Result { let mut gs_config_builder = gossipsub::ConfigBuilder::default(); gs_config_builder.max_transmit_size(1 << 20); @@ -109,13 +111,13 @@ pub(in crate::libp2p) fn build_gossipsub( let mut gossipsub = Gossipsub::new_with_subscription_filter( MessageAuthenticity::Signed(local_key.clone()), gossipsub_config, - build_subscription_filter(network_name), + build_subscription_filter(cfg), ) .map_err(anyhow::Error::msg)?; gossipsub .with_peer_score( - build_peer_score_params(network_name), + build_peer_score_params(cfg), build_peer_score_threshold(), ) .map_err(anyhow::Error::msg)?; @@ -128,6 +130,7 @@ impl ForestBehaviour { local_key: &Keypair, config: &Libp2pConfig, network_name: &GenesisNetworkName, + gossipsub: Gossipsub, peer_manager: Arc, ) -> anyhow::Result { const MAX_ESTABLISHED_PER_PEER: u32 = 4; @@ -146,8 +149,6 @@ impl ForestBehaviour { let max_concurrent_request_response_streams = (config.target_peer_count as usize) .saturating_mul(*MAX_CONCURRENT_REQUEST_RESPONSE_STREAMS_PER_PEER); - let gossipsub = build_gossipsub(local_key, network_name)?; - let bitswap = BitswapBehaviour::new( &[ "/chain/ipfs/bitswap/1.2.0", diff --git a/src/libp2p/gossip_params.rs b/src/libp2p/gossip_params.rs index 96a1ba54d1af..b8404e13e27f 100644 --- a/src/libp2p/gossip_params.rs +++ b/src/libp2p/gossip_params.rs @@ -7,9 +7,7 @@ use libp2p::gossipsub::{ PeerScoreParams, PeerScoreThresholds, TopicScoreParams, score_parameter_decay, }; -use strum::IntoEnumIterator as _; - -use crate::{libp2p::PubsubTopic, networks::GenesisNetworkName}; +use crate::{libp2p::{PubsubTopic, PubsubTopicCfg, pubsub_topics}}; // All these parameters are copied from what Lotus has set for their Topic // scores. They are currently unused because enabling them causes GossipSub @@ -81,17 +79,18 @@ fn build_block_topic_config() -> TopicScoreParams { } pub(in crate::libp2p) fn build_peer_score_params( - network_name: &GenesisNetworkName, + cfg: PubsubTopicCfg<'_> ) -> PeerScoreParams { #[allow(clippy::disallowed_types)] let mut psp_topics = std::collections::HashMap::new(); - for topic in PubsubTopic::iter() { - let params = match topic { + for (variant, topic) in pubsub_topics(cfg) { + let params = match variant { PubsubTopic::Blocks => build_block_topic_config(), PubsubTopic::Messages => build_msg_topic_config(), + PubsubTopic::Drand => Default::default(), }; - psp_topics.insert(topic.ident(network_name).hash(), params); + psp_topics.insert(topic.hash(), params); } PeerScoreParams { diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index f0ac4b257d07..2ecb5998ac4a 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -10,11 +10,13 @@ use crate::{ libp2p_bitswap::{BitswapStoreReadWrite, request_manager::BitswapRequestManager}, utils::flume::FlumeSenderExt as _, }; +use crate::beacon::PublicRandResponse; use crate::{message::SignedMessage, networks::GenesisNetworkName}; use ahash::{HashMap, HashSet}; use anyhow::Context as _; use flume::Sender; use futures::{select, stream::StreamExt as _}; +use libp2p::gossipsub::TopicHash; pub use libp2p::gossipsub::{IdentTopic, Topic}; use libp2p::{ PeerId, Swarm, SwarmBuilder, @@ -29,12 +31,15 @@ use libp2p::{ swarm::{DialError, SwarmEvent}, tcp, yamux, }; +use quick_protobuf::deserialize_from_slice; use nonzero_ext::nonzero; + use tokio_stream::wrappers::IntervalStream; use tracing::{debug, error, info, trace, warn}; use super::{ ForestBehaviour, ForestBehaviourEvent, Libp2pConfig, + behaviour::build_gossipsub, chain_exchange::{ChainExchangeRequest, ChainExchangeResponse, make_chain_exchange_response}, discovery::{DerivedDiscoveryBehaviourEvent, PeerInfo}, }; @@ -78,6 +83,8 @@ crate::def_is_env_truthy!(libp2p_metrics_enabled, "FOREST_LIBP2P_METRICS_ENABLED pub const PUBSUB_BLOCK_STR: &str = "/fil/blocks"; /// `Gossipsub` Filecoin messages topic identifier. pub const PUBSUB_MSG_STR: &str = "/fil/msgs"; +/// `Gossipsub` drand randomness topic identifier. +pub const PUBSUB_DRAND_STR: &str = "/drand/pubsub/v0.0.0"; /// Gossipsub topics Forest uses. Subscription, the subscription-filter /// whitelist, and peer-score params all iterate the variants, so adding one is @@ -88,21 +95,38 @@ pub enum PubsubTopic { Blocks, #[display("{PUBSUB_MSG_STR}")] Messages, + #[display("{PUBSUB_DRAND_STR}")] + Drand, } -impl PubsubTopic { - /// Full topic on `network_name`, e.g. `/fil/blocks/`. - pub fn ident(self, network_name: impl std::fmt::Display) -> IdentTopic { - IdentTopic::new(format!("{self}/{network_name}")) - } +#[derive(Clone, Copy)] +pub struct PubsubTopicCfg<'a> { + pub network_name: &'a GenesisNetworkName, + pub drand_chain_hashes: &'a [String], } /// All gossipsub topics on `network_name`. pub fn pubsub_topics( - network_name: impl std::fmt::Display + Copy, -) -> impl Iterator { + cfg: PubsubTopicCfg<'_>, +) -> Vec<(PubsubTopic, IdentTopic)> { use strum::IntoEnumIterator as _; - PubsubTopic::iter().map(move |t| t.ident(network_name)) + + let mut topics = Vec::new(); + for kind in PubsubTopic::iter() { + match kind { + PubsubTopic::Blocks | PubsubTopic::Messages => { + topics.push((kind, IdentTopic::new(format!("{kind}/{}", cfg.network_name)))); + }, + PubsubTopic::Drand => { + topics.extend( + cfg.drand_chain_hashes + .iter() + .map(move |h| (kind, IdentTopic::new(format!("{kind}/{h}")))) + ); + }, + } + } + topics } pub const BITSWAP_TIMEOUT: Duration = Duration::from_secs(30); @@ -137,6 +161,11 @@ pub enum PubsubMessage { Block(GossipBlock), /// Messages that come over the message topic Message(SignedMessage), + /// Messages that come over the drand topic + DrandEntry { + chain_hash: String, + response: PublicRandResponse, + }, } /// Messages into the service to handle. @@ -191,8 +220,8 @@ pub struct Libp2pService { network_sender_in: Sender, network_receiver_out: flume::Receiver, network_sender_out: Sender, - network_name: String, genesis_cid: Cid, + pubsub_topic_kinds: HashMap, } impl Libp2pService { @@ -204,9 +233,19 @@ impl Libp2pService { network_name: GenesisNetworkName, genesis_cid: Cid, ) -> anyhow::Result { - let behaviour = - ForestBehaviour::new(&net_keypair, &config, &network_name, peer_manager.clone()) - .await?; + let pubsub_topic_cfg = PubsubTopicCfg { + network_name: &network_name, + drand_chain_hashes: &cs.chain_config().drand_gossip_chain_hashes(), + }; + let gossipsub = build_gossipsub(&net_keypair, pubsub_topic_cfg)?; + let behaviour = ForestBehaviour::new( + &net_keypair, + &config, + &network_name, + gossipsub, + peer_manager.clone(), + ) + .await?; let mut swarm = SwarmBuilder::with_existing_identity(net_keypair) .with_tokio() .with_tcp( @@ -227,11 +266,14 @@ impl Libp2pService { .build(); // Subscribe to gossipsub topics with the network name suffix - for topic in pubsub_topics(&network_name) { + // and for drand uses the current drand network hash + let mut pubsub_topic_kinds = HashMap::default(); + for (kind, topic) in pubsub_topics(pubsub_topic_cfg) { swarm .behaviour_mut() .subscribe(&topic) .with_context(|| format!("Failed to subscribe gossipsub topic {topic}"))?; + pubsub_topic_kinds.insert(topic.hash(), kind); } let (network_sender_in, network_receiver_in) = flume::unbounded(); @@ -282,8 +324,8 @@ impl Libp2pService { network_sender_in, network_receiver_out, network_sender_out, - network_name: network_name.into(), genesis_cid, + pubsub_topic_kinds, }) } @@ -302,8 +344,8 @@ impl Libp2pService { let mut network_stream = self.network_receiver_in.stream().fuse(); let mut interval = IntervalStream::new(tokio::time::interval(Duration::from_secs(15))).fuse(); - let pubsub_block_str = PubsubTopic::Blocks.ident(&self.network_name).to_string(); - let pubsub_msg_str = PubsubTopic::Messages.ident(&self.network_name).to_string(); + + let pubsub_topic_kinds = self.pubsub_topic_kinds; let (cx_response_tx, cx_response_rx) = flume::unbounded(); @@ -345,8 +387,7 @@ impl Libp2pService { &self.genesis_cid, &self.network_sender_out, cx_response_tx.clone(), - &pubsub_block_str, - &pubsub_msg_str,).await; + &pubsub_topic_kinds).await; }, None => { break; }, _ => { }, @@ -654,8 +695,7 @@ async fn handle_discovery_event( async fn handle_gossip_event( e: gossipsub::Event, network_sender_out: &Sender, - pubsub_block_str: &str, - pubsub_msg_str: &str, + pubsub_topic_kinds: &HashMap, ) { if let gossipsub::Event::Message { propagation_source: source, @@ -663,11 +703,12 @@ async fn handle_gossip_event( .. } = e { - let topic = message.topic.as_str(); + let topic = message.topic; let message = message.data; trace!("Got a Gossip Message from {:?}", source); - if topic == pubsub_block_str { - match from_slice_with_fallback::(&message) { + + match pubsub_topic_kinds.get(&topic) { + Some(PubsubTopic::Blocks) => match from_slice_with_fallback::(&message) { Ok(b) => { emit_event( network_sender_out, @@ -681,8 +722,7 @@ async fn handle_gossip_event( warn!("Gossip Block from peer {source:?} could not be deserialized: {e:#}",); } } - } else if topic == pubsub_msg_str { - match from_slice_with_fallback::(&message) { + Some(PubsubTopic::Messages) => match from_slice_with_fallback::(&message) { Ok(m) => { emit_event( network_sender_out, @@ -696,8 +736,35 @@ async fn handle_gossip_event( warn!("Gossip Message from peer {source:?} could not be deserialized: {e:#}"); } } - } else { - warn!("Getting gossip messages from unknown topic: {topic}"); + Some(PubsubTopic::Drand) => { + // `IdentTopic` hashes to its own string, so the chain hash is the topic suffix. + let Some(chain_hash) = topic + .as_str() + .strip_prefix(PUBSUB_DRAND_STR) + .and_then(|suffix| suffix.strip_prefix('/')) + else { + warn!("Malformed drand topic: {topic}"); + return; + }; + match deserialize_from_slice::(&message) { + Ok(response) => { + emit_event( + network_sender_out, + NetworkEvent::PubsubMessage { + message: PubsubMessage::DrandEntry { + chain_hash: chain_hash.to_string(), + response, + }, + }, + ) + .await; + } + Err(e) => { + warn!("Gossip drand entry from peer {source:?} could not be decoded: {e:#}"); + } + } + } + None => warn!("Getting gossip messages from unknown topic: {topic}"), } } } @@ -923,8 +990,7 @@ async fn handle_forest_behaviour_event( request_response::ResponseChannel, ChainExchangeResponse, )>, - pubsub_block_str: &str, - pubsub_msg_str: &str, + pubsub_topic_kinds: &HashMap, ) { match event { ForestBehaviourEvent::Discovery(discovery_out) => { @@ -937,7 +1003,7 @@ async fn handle_forest_behaviour_event( .await } ForestBehaviourEvent::Gossipsub(e) => { - handle_gossip_event(e, network_sender_out, pubsub_block_str, pubsub_msg_str).await + handle_gossip_event(e, network_sender_out, pubsub_topic_kinds).await } ForestBehaviourEvent::Hello(rr_event) => { let behaviour_mut = swarm.behaviour_mut(); diff --git a/src/networks/mod.rs b/src/networks/mod.rs index 08d1fc37267c..626a3de63a2d 100644 --- a/src/networks/mod.rs +++ b/src/networks/mod.rs @@ -3,6 +3,7 @@ use std::str::FromStr; use std::sync::LazyLock; +use std::slice::Iter; use indexmap::IndexMap; use libp2p::Multiaddr; @@ -486,16 +487,25 @@ impl ChainConfig { 0 } - pub fn get_beacon_schedule(&self, genesis_ts: u64) -> BeaconSchedule { - let ds_iter = match self.network { + fn drand_points(&self) -> Iter<'_, DrandPoint<'static>> { + match self.network { NetworkChain::Mainnet => mainnet::DRAND_SCHEDULE.iter(), NetworkChain::Calibnet => calibnet::DRAND_SCHEDULE.iter(), NetworkChain::Butterflynet => butterflynet::DRAND_SCHEDULE.iter(), NetworkChain::Devnet(_) => devnet::DRAND_SCHEDULE.iter(), - }; + } + } + + pub fn drand_gossip_chain_hashes(&self) -> Vec { + self.drand_points() + .filter(|p| p.config.network_type.is_unchained()) + .map(|p| p.config.chain_info.hash.to_string()) + .collect() + } + pub fn get_beacon_schedule(&self, genesis_ts: u64) -> BeaconSchedule { BeaconSchedule( - ds_iter + self.drand_points() .map(|dc| { BeaconPoint::new( dc.height, From f113fa50c613b86dc985128ec40a134892642421 Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Mon, 24 Aug 2026 11:14:39 -0400 Subject: [PATCH 02/12] chore: implement gosssip subscription + watchdog --- src/beacon/drand.rs | 43 ++++++-- src/beacon/metrics.rs | 43 ++++++++ src/beacon/mod.rs | 1 + src/beacon/tests/drand.rs | 33 +++++++ src/chain_sync/chain_follower.rs | 113 ++++++++++++++++++++-- src/libp2p/behaviour.rs | 11 +-- src/libp2p/gossip_params.rs | 6 +- src/libp2p/service.rs | 79 +++++++-------- src/libp2p/tests/gossipsub_filter_test.rs | 54 +++++++++-- src/networks/mod.rs | 14 ++- 10 files changed, 316 insertions(+), 81 deletions(-) create mode 100644 src/beacon/metrics.rs diff --git a/src/beacon/drand.rs b/src/beacon/drand.rs index 887b78d77591..ff1bdea74350 100644 --- a/src/beacon/drand.rs +++ b/src/beacon/drand.rs @@ -14,6 +14,8 @@ use super::{ PublicKeyOnG1, PublicKeyOnG2, SignatureOnG1, SignatureOnG2, verify_messages_chained, }, }; +use crate::beacon::metrics; +use crate::metrics::HistogramTimerExt as _; use crate::prelude::*; use crate::shim::clock::ChainEpoch; use crate::shim::version::NetworkVersion; @@ -37,7 +39,8 @@ pub static IGNORE_DRAND: LazyLock = LazyLock::new(|| is_env_truthy(IGNORE_ /// Type of the `drand` network. `mainnet` is chained and `quicknet` is unchained. /// For the details, see -#[derive(PartialEq, Eq, Copy, Clone, Debug, SerdeSerialize, SerdeDeserialize)] +#[derive(PartialEq, Eq, Copy, Clone, Debug, SerdeSerialize, SerdeDeserialize, strum::Display)] +#[strum(serialize_all = "snake_case")] pub enum DrandNetwork { Mainnet, Quicknet, @@ -135,6 +138,13 @@ impl BeaconSchedule { } } + pub fn unchained_beacon(&self) -> Option<&BeaconImpl> { + self.0 + .iter() + .map(|point| &point.beacon) + .find(|beacon| beacon.network().is_unchained()) + } + pub fn beacon_for_epoch(&self, epoch: ChainEpoch) -> anyhow::Result<(ChainEpoch, &BeaconImpl)> { // Iterate over beacon schedule to find the latest randomness beacon to use. self.0 @@ -271,7 +281,10 @@ impl DrandBeacon { drand_gen_time: config.chain_info.genesis_time as u64, fil_round_time: interval, fil_gen_time: genesis_ts, - verified_beacons: SizeTrackingCache::new_with_metrics("verified_beacons", CACHE_SIZE), + verified_beacons: SizeTrackingCache::new_with_metrics( + format!("verified_beacons_{}", config.network_type), + CACHE_SIZE, + ), } } @@ -371,9 +384,14 @@ impl Beacon for DrandBeacon { async fn entry(&self, round: u64) -> anyhow::Result { if let Some(cached_entry) = self.verified_beacons.get(&round) { + metrics::DRAND_ENTRY_SOURCE_TOTAL + .get_or_create(&metrics::CACHE) + .inc(); return Ok(Arc::unwrap_or_clone(cached_entry)); } + let _timer = metrics::DRAND_HTTP_FETCH_TIME.start_timer(); + async fn fetch_entry_from_url(url: impl reqwest::IntoUrl) -> anyhow::Result { let resp: BeaconEntryJson = global_http_client() .get(url) @@ -416,16 +434,25 @@ impl Beacon for DrandBeacon { humantime::format_duration(dur) ); }) - .await?; + .await + .inspect_err(|_| { + metrics::DRAND_ENTRY_SOURCE_TOTAL + .get_or_create(&metrics::HTTP_ERROR) + .inc(); + })?; // Callers assume the entry is for the round they asked for. Round 0 is served // as "latest", so it answers with a different round by design: // - anyhow::ensure!( - round == 0 || entry.round() == round, - "drand returned round {} for round {round}", - entry.round() - ); + if round != 0 && entry.round() != round { + metrics::DRAND_ENTRY_SOURCE_TOTAL + .get_or_create(&metrics::HTTP_ERROR) + .inc(); + anyhow::bail!("drand returned round {} for round {round}", entry.round()); + } self.cache_fetched_entry(&entry); + metrics::DRAND_ENTRY_SOURCE_TOTAL + .get_or_create(&metrics::HTTP) + .inc(); Ok(entry) } diff --git a/src/beacon/metrics.rs b/src/beacon/metrics.rs new file mode 100644 index 000000000000..4e3f4a6cb449 --- /dev/null +++ b/src/beacon/metrics.rs @@ -0,0 +1,43 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use prometheus_client::{ + encoding::EncodeLabelSet, + metrics::{counter::Counter, family::Family, histogram::Histogram}, +}; +use std::sync::LazyLock; + +#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet, derive_more::Constructor)] +pub struct DrandSourceLabel { + pub source: &'static str, +} + +// drand_entry_source_total +pub const CACHE: DrandSourceLabel = DrandSourceLabel::new("cache"); +pub const HTTP: DrandSourceLabel = DrandSourceLabel::new("http"); +pub const HTTP_ERROR: DrandSourceLabel = DrandSourceLabel::new("http_error"); + +/// Counts every round served by [`crate::beacon::Beacon::entry`], labelled by where it +/// came from. +pub static DRAND_ENTRY_SOURCE_TOTAL: LazyLock> = + LazyLock::new(|| { + let metric = Family::default(); + crate::metrics::default_registry().register( + "drand_entry_source_total", + "Total number of drand rounds served, by source", + metric.clone(), + ); + metric + }); + +/// Wall-clock duration of a drand HTTP fetch, covering the whole retry chain across +/// every configured server rather than a single attempt. +pub static DRAND_HTTP_FETCH_TIME: LazyLock = LazyLock::new(|| { + let metric = crate::metrics::default_histogram(); + crate::metrics::default_registry().register( + "drand_http_fetch_time", + "Duration of a drand HTTP round fetch, including retries across servers", + metric.clone(), + ); + metric +}); diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index b4b536c4f5d4..2eb08507da9f 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -4,6 +4,7 @@ pub mod beacon_entries; mod drand; mod drand_pb; +pub mod metrics; pub mod signatures; pub use beacon_entries::*; pub use drand::*; diff --git a/src/beacon/tests/drand.rs b/src/beacon/tests/drand.rs index b5611b30fbf8..f5a34e3658a8 100644 --- a/src/beacon/tests/drand.rs +++ b/src/beacon/tests/drand.rs @@ -308,3 +308,36 @@ async fn beacon_entries_for_block_covers_null_rounds_quicknet() { ); } } + +#[test] +#[serial_test::serial] +fn verified_beacons_cache_metrics_are_uniquely_named() { + use crate::networks::ChainConfig; + + crate::metrics::reset_collector_registry(); + // Mainnet has three drand points: Incentinet, Mainnet and Quicknet. + let schedule = ChainConfig::mainnet().get_beacon_schedule(1598306400); + assert_eq!(schedule.0.len(), 3); + + let mut encoded = String::new(); + prometheus_client::encoding::text::encode_registry( + &mut encoded, + &crate::metrics::collector_registry(), + ) + .unwrap(); + + let families: Vec<_> = encoded + .lines() + .filter_map(|line| line.strip_prefix("# HELP ")) + .filter(|line| line.starts_with("cache_verified_beacons")) + .map(|line| line.split_whitespace().next().unwrap_or_default()) + .collect(); + + // Five metrics (size/len/cap/hits/misses) for each of the three beacons. + assert_eq!(families.len(), 15); + assert_eq!( + families.iter().unique().count(), + families.len(), + "duplicate cache metric families: {families:?}" + ); +} diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index d79c89d91de2..ce02a0522648 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -18,6 +18,7 @@ use super::network_context::SyncNetworkContext; use crate::{ + beacon::{Beacon, BeaconEntry}, blocks::{Block, FullTipset, Tipset, TipsetKey}, chain::{ChainStore, index::ResolveNullTipset}, chain_sync::{ @@ -41,6 +42,7 @@ use hashbrown::{HashMap, HashSet}; use libp2p::PeerId; use nonzero_ext::nonzero; use parking_lot::Mutex; +use std::sync::atomic::{AtomicU64, Ordering}; use std::{ borrow::Cow, sync::LazyLock, @@ -223,6 +225,8 @@ async fn chain_follower( let hello_fetch_limiter = Arc::new(Semaphore::new(*MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES)); + let last_drand_entry = Arc::new(AtomicU64::new(0)); + let mut set = JoinSet::new(); let cancellation_token = CancellationToken::new(); let _cancellation_token_drop_guard = cancellation_token.drop_guard_ref(); @@ -238,6 +242,7 @@ async fn chain_follower( let cancellation_token = cancellation_token.clone(); let hello_fetch_limiter = hello_fetch_limiter.shallow_clone(); let tipset_sender = tipset_sender.clone(); + let last_drand_entry = last_drand_entry.clone(); async move { while let Ok(event) = network_rx.recv_async().await { inc_gossipsub_event_metrics(&event); @@ -304,16 +309,35 @@ async fn chain_follower( debug!("Received invalid GossipSub message: {}", why); } } - PubsubMessage::DrandEntry { - chain_hash, - response, - } => { - // Signature verification and cache insertion land with the - // beacon-schedule wiring. - tracing::trace!( - "Received drand round {} on chain {chain_hash}", - response.round - ); + PubsubMessage::DrandEntry(entry) => { + if entry.round() == 0 || entry.signature().is_empty() { + continue; + } + let beacon_schedule = state_manager.beacon_schedule().clone(); + let last_drand_entry = last_drand_entry.clone(); + tokio::task::spawn_blocking(move || { + let Some(beacon) = beacon_schedule.unchained_beacon() else { + return; + }; + + if matches!( + beacon.verify_entries( + std::slice::from_ref(&entry), + &BeaconEntry::default() + ), + Ok(true) + ) { + last_drand_entry.store( + Utc::now().timestamp().max(0) as u64, + Ordering::Relaxed, + ); + } else { + debug!( + round = entry.round(), + "received invalid drand entry over gossipsub" + ); + } + }); } }, _ => {} @@ -322,6 +346,15 @@ async fn chain_follower( } }); + set.spawn({ + let state_manager = state_manager.shallow_clone(); + let last_drand_entry = last_drand_entry.clone(); + let cancellation_token = cancellation_token.clone(); + async move { + drand_gossip_watchdog(state_manager, last_drand_entry, cancellation_token).await; + } + }); + // Forward tipsets from miners into the state machine. set.spawn({ let state_changed = state_changed.clone(); @@ -479,6 +512,66 @@ async fn chain_follower( Ok(()) } +/// Watch the drand gossipsub for staleness, if a drand breacon entry +/// is not received in half a chain epoch then we consider it stale for +/// that epoch and fallback to fetch the beacon through HTTP +async fn drand_gossip_watchdog( + state_manager: StateManager, + last_drand_entry: Arc, + cancellation_token: CancellationToken, +) { + let chain_config = state_manager.chain_config(); + let Some(period_secs) = chain_config.drand_gossip_period_secs() else { + return; + }; + + let deadline = Duration::from_secs(period_secs.div_ceil(2)); + + let mut ticker = tokio::time::interval(deadline); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut stale = false; + + while cancellation_token + .run_until_cancelled(ticker.tick()) + .await + .is_some() + { + let last_seen = last_drand_entry.load(Ordering::Relaxed); + let now = Utc::now().timestamp().max(0) as u64; + if last_seen != 0 && now.saturating_sub(last_seen) < deadline.as_secs() { + if stale { + stale = false; + info!("drand gossipsub entries are flowing again"); + } + continue; + } + + if !stale { + stale = true; + warn!( + deadline_secs = deadline.as_secs(), + "no verified drand entry over gossipsub within the deadline, falling back to HTTP" + ); + } + + let Some(beacon) = state_manager.beacon_schedule().unchained_beacon() else { + continue; + }; + let epoch = state_manager.heaviest_tipset().epoch() + 1; + let network_version = state_manager.get_network_version(epoch); + let round = match beacon.max_beacon_round_for_epoch(network_version, epoch) { + Ok(round) => round, + Err(e) => { + debug!("no drand round for epoch {epoch}: {e:#}"); + continue; + } + }; + if let Err(e) = beacon.entry(round).await { + debug!("drand HTTP fallback for round {round} failed: {e:#}"); + } + } +} + // Increment the gossipsub event metrics. fn inc_gossipsub_event_metrics(event: &NetworkEvent) { let label = match event { diff --git a/src/libp2p/behaviour.rs b/src/libp2p/behaviour.rs index 3a2b36aba095..37cf9df89af6 100644 --- a/src/libp2p/behaviour.rs +++ b/src/libp2p/behaviour.rs @@ -10,8 +10,8 @@ use super::{ PeerManager, discovery::{DerivedDiscoveryBehaviourEvent, DiscoveryEvent, PeerInfo}, }; -use crate::{libp2p::PubsubTopicCfg, libp2p_bitswap::BitswapBehaviour}; use crate::utils::{encoding::blake2b_256, version::FOREST_VERSION_STRING}; +use crate::{libp2p::PubsubTopicCfg, libp2p_bitswap::BitswapBehaviour}; use crate::{ libp2p::{ chain_exchange::ChainExchangeBehaviour, @@ -79,7 +79,7 @@ const MAX_SUBSCRIPTIONS_PER_REQUEST: usize = 100; /// Filter accepting only Forest's topics, bounded in count and per request. pub(in crate::libp2p) fn build_subscription_filter( - cfg: PubsubTopicCfg<'_> + cfg: PubsubTopicCfg<'_>, ) -> MaxCountSubscriptionFilter { let allowed: Vec<_> = crate::libp2p::pubsub_topics(cfg) .iter() @@ -97,7 +97,7 @@ pub(in crate::libp2p) fn build_subscription_filter( pub(in crate::libp2p) fn build_gossipsub( local_key: &Keypair, - cfg: PubsubTopicCfg<'_> + cfg: PubsubTopicCfg<'_>, ) -> anyhow::Result { let mut gs_config_builder = gossipsub::ConfigBuilder::default(); gs_config_builder.max_transmit_size(1 << 20); @@ -116,10 +116,7 @@ pub(in crate::libp2p) fn build_gossipsub( .map_err(anyhow::Error::msg)?; gossipsub - .with_peer_score( - build_peer_score_params(cfg), - build_peer_score_threshold(), - ) + .with_peer_score(build_peer_score_params(cfg), build_peer_score_threshold()) .map_err(anyhow::Error::msg)?; Ok(gossipsub) diff --git a/src/libp2p/gossip_params.rs b/src/libp2p/gossip_params.rs index b8404e13e27f..3da6dcc92d1b 100644 --- a/src/libp2p/gossip_params.rs +++ b/src/libp2p/gossip_params.rs @@ -7,7 +7,7 @@ use libp2p::gossipsub::{ PeerScoreParams, PeerScoreThresholds, TopicScoreParams, score_parameter_decay, }; -use crate::{libp2p::{PubsubTopic, PubsubTopicCfg, pubsub_topics}}; +use crate::libp2p::{PubsubTopic, PubsubTopicCfg, pubsub_topics}; // All these parameters are copied from what Lotus has set for their Topic // scores. They are currently unused because enabling them causes GossipSub @@ -78,9 +78,7 @@ fn build_block_topic_config() -> TopicScoreParams { } } -pub(in crate::libp2p) fn build_peer_score_params( - cfg: PubsubTopicCfg<'_> -) -> PeerScoreParams { +pub(in crate::libp2p) fn build_peer_score_params(cfg: PubsubTopicCfg<'_>) -> PeerScoreParams { #[allow(clippy::disallowed_types)] let mut psp_topics = std::collections::HashMap::new(); diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index 2ecb5998ac4a..3ca584162eec 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -3,6 +3,7 @@ use std::time::{Duration, UNIX_EPOCH}; +use crate::beacon::{BeaconEntry, PublicRandResponse}; use crate::prelude::*; use crate::{blocks::GossipBlock, rpc::net::NetInfoResult}; use crate::{chain::ChainStore, utils::encoding::from_slice_with_fallback}; @@ -10,7 +11,6 @@ use crate::{ libp2p_bitswap::{BitswapStoreReadWrite, request_manager::BitswapRequestManager}, utils::flume::FlumeSenderExt as _, }; -use crate::beacon::PublicRandResponse; use crate::{message::SignedMessage, networks::GenesisNetworkName}; use ahash::{HashMap, HashSet}; use anyhow::Context as _; @@ -31,8 +31,8 @@ use libp2p::{ swarm::{DialError, SwarmEvent}, tcp, yamux, }; -use quick_protobuf::deserialize_from_slice; use nonzero_ext::nonzero; +use quick_protobuf::deserialize_from_slice; use tokio_stream::wrappers::IntervalStream; use tracing::{debug, error, info, trace, warn}; @@ -106,24 +106,25 @@ pub struct PubsubTopicCfg<'a> { } /// All gossipsub topics on `network_name`. -pub fn pubsub_topics( - cfg: PubsubTopicCfg<'_>, -) -> Vec<(PubsubTopic, IdentTopic)> { +pub fn pubsub_topics(cfg: PubsubTopicCfg<'_>) -> Vec<(PubsubTopic, IdentTopic)> { use strum::IntoEnumIterator as _; let mut topics = Vec::new(); for kind in PubsubTopic::iter() { match kind { PubsubTopic::Blocks | PubsubTopic::Messages => { - topics.push((kind, IdentTopic::new(format!("{kind}/{}", cfg.network_name)))); - }, + topics.push(( + kind, + IdentTopic::new(format!("{kind}/{}", cfg.network_name)), + )); + } PubsubTopic::Drand => { topics.extend( cfg.drand_chain_hashes .iter() - .map(move |h| (kind, IdentTopic::new(format!("{kind}/{h}")))) + .map(move |h| (kind, IdentTopic::new(format!("{kind}/{h}")))), ); - }, + } } } topics @@ -162,10 +163,7 @@ pub enum PubsubMessage { /// Messages that come over the message topic Message(SignedMessage), /// Messages that come over the drand topic - DrandEntry { - chain_hash: String, - response: PublicRandResponse, - }, + DrandEntry(BeaconEntry), } /// Messages into the service to handle. @@ -721,50 +719,47 @@ async fn handle_gossip_event( Err(e) => { warn!("Gossip Block from peer {source:?} could not be deserialized: {e:#}",); } - } - Some(PubsubTopic::Messages) => match from_slice_with_fallback::(&message) { - Ok(m) => { - emit_event( - network_sender_out, - NetworkEvent::PubsubMessage { - message: PubsubMessage::Message(m), - }, - ) - .await; - } - Err(e) => { - warn!("Gossip Message from peer {source:?} could not be deserialized: {e:#}"); + }, + Some(PubsubTopic::Messages) => { + match from_slice_with_fallback::(&message) { + Ok(m) => { + emit_event( + network_sender_out, + NetworkEvent::PubsubMessage { + message: PubsubMessage::Message(m), + }, + ) + .await; + } + Err(e) => { + warn!( + "Gossip Message from peer {source:?} could not be deserialized: {e:#}" + ); + } } } Some(PubsubTopic::Drand) => { - // `IdentTopic` hashes to its own string, so the chain hash is the topic suffix. - let Some(chain_hash) = topic - .as_str() - .strip_prefix(PUBSUB_DRAND_STR) - .and_then(|suffix| suffix.strip_prefix('/')) - else { - warn!("Malformed drand topic: {topic}"); - return; - }; match deserialize_from_slice::(&message) { - Ok(response) => { + Ok(r) => { emit_event( network_sender_out, NetworkEvent::PubsubMessage { - message: PubsubMessage::DrandEntry { - chain_hash: chain_hash.to_string(), - response, - }, + message: PubsubMessage::DrandEntry(BeaconEntry::new( + r.round, + r.signature, + )), }, ) .await; } Err(e) => { - warn!("Gossip drand entry from peer {source:?} could not be decoded: {e:#}"); + warn!( + "Gossip drand entry from peer {source:?} could not be decoded: {e:#}" + ); } } } - None => warn!("Getting gossip messages from unknown topic: {topic}"), + None => warn!("Getting gossip messages from unknown topic: {topic}"), } } } diff --git a/src/libp2p/tests/gossipsub_filter_test.rs b/src/libp2p/tests/gossipsub_filter_test.rs index 8b0d05ed0eec..81e6e6a5830c 100644 --- a/src/libp2p/tests/gossipsub_filter_test.rs +++ b/src/libp2p/tests/gossipsub_filter_test.rs @@ -14,14 +14,51 @@ use libp2p::{ }; use libp2p_swarm_test::SwarmExt as _; -use crate::libp2p::{Gossipsub, build_gossipsub, build_subscription_filter, pubsub_topics}; +use crate::libp2p::{ + Gossipsub, PubsubTopicCfg, build_gossipsub, build_subscription_filter, pubsub_topics, +}; +use crate::networks::GenesisNetworkName; const NETWORK: &str = "testnetname"; +/// quicknet, the one unchained drand network Forest subscribes to. +const DRAND_HASH: &str = "52db9ba70e0cc0f6eaf7803dd07447a1f5477735fd3f661792ba94600c84e971"; + +/// Owns what [`PubsubTopicCfg`] borrows. +struct TopicCfgOwner { + network_name: GenesisNetworkName, + drand_chain_hashes: Vec, +} + +impl TopicCfgOwner { + fn new() -> Self { + Self { + network_name: NETWORK.into(), + drand_chain_hashes: vec![DRAND_HASH.to_string()], + } + } + + fn cfg(&self) -> PubsubTopicCfg<'_> { + PubsubTopicCfg { + network_name: &self.network_name, + drand_chain_hashes: &self.drand_chain_hashes, + } + } +} + +/// Every topic the node should accept, drand included. +fn allowed_topics() -> Vec { + let owner = TopicCfgOwner::new(); + pubsub_topics(owner.cfg()) + .into_iter() + .map(|(_, topic)| topic) + .collect() +} /// Swarm using Forest's subscription filter (the code under test). fn filtered_swarm() -> Swarm { Swarm::new_ephemeral_tokio(|identity| { - build_gossipsub(&identity, &NETWORK.into()).expect("failed to build gossipsub") + let owner = TopicCfgOwner::new(); + build_gossipsub(&identity, owner.cfg()).expect("failed to build gossipsub") }) } @@ -53,7 +90,7 @@ async fn only_whitelisted_topics_are_tracked() { let unlisted = IdentTopic::new(format!("/other/topic/{i}")); peer.behaviour_mut().subscribe(&unlisted).unwrap(); } - let allowed: Vec = pubsub_topics(NETWORK).collect(); + let allowed = allowed_topics(); for topic in &allowed { peer.behaviour_mut().subscribe(topic).unwrap(); } @@ -84,19 +121,22 @@ async fn only_whitelisted_topics_are_tracked() { #[test] fn filter_allows_only_whitelisted_topics() { - let mut filter = build_subscription_filter(&NETWORK.into()); - for topic in pubsub_topics(NETWORK) { + let owner = TopicCfgOwner::new(); + let mut filter = build_subscription_filter(owner.cfg()); + for topic in allowed_topics() { assert!(filter.can_subscribe(&topic.hash())); } assert!(!filter.can_subscribe(&IdentTopic::new("/cth/ulhu").hash())); assert!(!filter.can_subscribe(&TopicHash::from_raw("x".repeat(1 << 20)))); // Wrong network suffix must not match. assert!(!filter.can_subscribe(&IdentTopic::new("/fil/blocks/lovecraftnet").hash())); + assert!(!filter.can_subscribe(&IdentTopic::new("/drand/pubsub/v0.0.0/deadbeef").hash())); } #[test] fn filter_caps_are_set() { - let filter = build_subscription_filter(&NETWORK.into()); - assert_eq!(filter.max_subscribed_topics, pubsub_topics(NETWORK).count()); + let owner = TopicCfgOwner::new(); + let filter = build_subscription_filter(owner.cfg()); + assert_eq!(filter.max_subscribed_topics, allowed_topics().len()); assert_eq!(filter.max_subscriptions_per_request, 100); } diff --git a/src/networks/mod.rs b/src/networks/mod.rs index 626a3de63a2d..a8e9eaeace7a 100644 --- a/src/networks/mod.rs +++ b/src/networks/mod.rs @@ -1,9 +1,9 @@ // Copyright 2019-2026 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT +use std::slice::Iter; use std::str::FromStr; use std::sync::LazyLock; -use std::slice::Iter; use indexmap::IndexMap; use libp2p::Multiaddr; @@ -487,7 +487,7 @@ impl ChainConfig { 0 } - fn drand_points(&self) -> Iter<'_, DrandPoint<'static>> { + fn drand_points(&self) -> Iter<'_, DrandPoint<'static>> { match self.network { NetworkChain::Mainnet => mainnet::DRAND_SCHEDULE.iter(), NetworkChain::Calibnet => calibnet::DRAND_SCHEDULE.iter(), @@ -495,7 +495,7 @@ impl ChainConfig { NetworkChain::Devnet(_) => devnet::DRAND_SCHEDULE.iter(), } } - + pub fn drand_gossip_chain_hashes(&self) -> Vec { self.drand_points() .filter(|p| p.config.network_type.is_unchained()) @@ -503,6 +503,14 @@ impl ChainConfig { .collect() } + /// Round interval, in seconds, of the drand network Forest subscribes to over + /// gossipsub. `None` when the network has no unchained drand point. + pub fn drand_gossip_period_secs(&self) -> Option { + self.drand_points() + .find(|p| p.config.network_type.is_unchained()) + .map(|p| p.config.chain_info.period as u64) + } + pub fn get_beacon_schedule(&self, genesis_ts: u64) -> BeaconSchedule { BeaconSchedule( self.drand_points() From 360d5424aed69ac1d4753c03d9fa78dd6579b72c Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Mon, 24 Aug 2026 17:14:55 -0400 Subject: [PATCH 03/12] chore: rever the metrics changes --- src/beacon/drand.rs | 36 ++++++++---------------------------- src/beacon/metrics.rs | 43 ------------------------------------------- src/beacon/mod.rs | 1 - 3 files changed, 8 insertions(+), 72 deletions(-) delete mode 100644 src/beacon/metrics.rs diff --git a/src/beacon/drand.rs b/src/beacon/drand.rs index ff1bdea74350..ce56ecf7c1c0 100644 --- a/src/beacon/drand.rs +++ b/src/beacon/drand.rs @@ -14,8 +14,6 @@ use super::{ PublicKeyOnG1, PublicKeyOnG2, SignatureOnG1, SignatureOnG2, verify_messages_chained, }, }; -use crate::beacon::metrics; -use crate::metrics::HistogramTimerExt as _; use crate::prelude::*; use crate::shim::clock::ChainEpoch; use crate::shim::version::NetworkVersion; @@ -39,8 +37,7 @@ pub static IGNORE_DRAND: LazyLock = LazyLock::new(|| is_env_truthy(IGNORE_ /// Type of the `drand` network. `mainnet` is chained and `quicknet` is unchained. /// For the details, see -#[derive(PartialEq, Eq, Copy, Clone, Debug, SerdeSerialize, SerdeDeserialize, strum::Display)] -#[strum(serialize_all = "snake_case")] +#[derive(PartialEq, Eq, Copy, Clone, Debug, SerdeSerialize, SerdeDeserialize)] pub enum DrandNetwork { Mainnet, Quicknet, @@ -281,10 +278,7 @@ impl DrandBeacon { drand_gen_time: config.chain_info.genesis_time as u64, fil_round_time: interval, fil_gen_time: genesis_ts, - verified_beacons: SizeTrackingCache::new_with_metrics( - format!("verified_beacons_{}", config.network_type), - CACHE_SIZE, - ), + verified_beacons: SizeTrackingCache::new_with_metrics("verified_beacons", CACHE_SIZE), } } @@ -384,14 +378,9 @@ impl Beacon for DrandBeacon { async fn entry(&self, round: u64) -> anyhow::Result { if let Some(cached_entry) = self.verified_beacons.get(&round) { - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::CACHE) - .inc(); return Ok(Arc::unwrap_or_clone(cached_entry)); } - let _timer = metrics::DRAND_HTTP_FETCH_TIME.start_timer(); - async fn fetch_entry_from_url(url: impl reqwest::IntoUrl) -> anyhow::Result { let resp: BeaconEntryJson = global_http_client() .get(url) @@ -434,25 +423,16 @@ impl Beacon for DrandBeacon { humantime::format_duration(dur) ); }) - .await - .inspect_err(|_| { - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::HTTP_ERROR) - .inc(); - })?; + .await?; // Callers assume the entry is for the round they asked for. Round 0 is served // as "latest", so it answers with a different round by design: // - if round != 0 && entry.round() != round { - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::HTTP_ERROR) - .inc(); - anyhow::bail!("drand returned round {} for round {round}", entry.round()); - } + anyhow::ensure!( + round == 0 || entry.round() == round, + "drand returned round {} for round {round}", + entry.round() + ); self.cache_fetched_entry(&entry); - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::HTTP) - .inc(); Ok(entry) } diff --git a/src/beacon/metrics.rs b/src/beacon/metrics.rs deleted file mode 100644 index 4e3f4a6cb449..000000000000 --- a/src/beacon/metrics.rs +++ /dev/null @@ -1,43 +0,0 @@ -// Copyright 2019-2026 ChainSafe Systems -// SPDX-License-Identifier: Apache-2.0, MIT - -use prometheus_client::{ - encoding::EncodeLabelSet, - metrics::{counter::Counter, family::Family, histogram::Histogram}, -}; -use std::sync::LazyLock; - -#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet, derive_more::Constructor)] -pub struct DrandSourceLabel { - pub source: &'static str, -} - -// drand_entry_source_total -pub const CACHE: DrandSourceLabel = DrandSourceLabel::new("cache"); -pub const HTTP: DrandSourceLabel = DrandSourceLabel::new("http"); -pub const HTTP_ERROR: DrandSourceLabel = DrandSourceLabel::new("http_error"); - -/// Counts every round served by [`crate::beacon::Beacon::entry`], labelled by where it -/// came from. -pub static DRAND_ENTRY_SOURCE_TOTAL: LazyLock> = - LazyLock::new(|| { - let metric = Family::default(); - crate::metrics::default_registry().register( - "drand_entry_source_total", - "Total number of drand rounds served, by source", - metric.clone(), - ); - metric - }); - -/// Wall-clock duration of a drand HTTP fetch, covering the whole retry chain across -/// every configured server rather than a single attempt. -pub static DRAND_HTTP_FETCH_TIME: LazyLock = LazyLock::new(|| { - let metric = crate::metrics::default_histogram(); - crate::metrics::default_registry().register( - "drand_http_fetch_time", - "Duration of a drand HTTP round fetch, including retries across servers", - metric.clone(), - ); - metric -}); diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index 2eb08507da9f..b4b536c4f5d4 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -4,7 +4,6 @@ pub mod beacon_entries; mod drand; mod drand_pb; -pub mod metrics; pub mod signatures; pub use beacon_entries::*; pub use drand::*; From e5c47e3fb65b76e11e8f14d524e2f0e463773ea8 Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Mon, 24 Aug 2026 17:14:55 -0400 Subject: [PATCH 04/12] chore: rever the metrics changes --- src/beacon/drand.rs | 36 ++++++++------------------------ src/beacon/metrics.rs | 43 --------------------------------------- src/beacon/mod.rs | 1 - src/beacon/tests/drand.rs | 33 ------------------------------ 4 files changed, 8 insertions(+), 105 deletions(-) delete mode 100644 src/beacon/metrics.rs diff --git a/src/beacon/drand.rs b/src/beacon/drand.rs index ff1bdea74350..ce56ecf7c1c0 100644 --- a/src/beacon/drand.rs +++ b/src/beacon/drand.rs @@ -14,8 +14,6 @@ use super::{ PublicKeyOnG1, PublicKeyOnG2, SignatureOnG1, SignatureOnG2, verify_messages_chained, }, }; -use crate::beacon::metrics; -use crate::metrics::HistogramTimerExt as _; use crate::prelude::*; use crate::shim::clock::ChainEpoch; use crate::shim::version::NetworkVersion; @@ -39,8 +37,7 @@ pub static IGNORE_DRAND: LazyLock = LazyLock::new(|| is_env_truthy(IGNORE_ /// Type of the `drand` network. `mainnet` is chained and `quicknet` is unchained. /// For the details, see -#[derive(PartialEq, Eq, Copy, Clone, Debug, SerdeSerialize, SerdeDeserialize, strum::Display)] -#[strum(serialize_all = "snake_case")] +#[derive(PartialEq, Eq, Copy, Clone, Debug, SerdeSerialize, SerdeDeserialize)] pub enum DrandNetwork { Mainnet, Quicknet, @@ -281,10 +278,7 @@ impl DrandBeacon { drand_gen_time: config.chain_info.genesis_time as u64, fil_round_time: interval, fil_gen_time: genesis_ts, - verified_beacons: SizeTrackingCache::new_with_metrics( - format!("verified_beacons_{}", config.network_type), - CACHE_SIZE, - ), + verified_beacons: SizeTrackingCache::new_with_metrics("verified_beacons", CACHE_SIZE), } } @@ -384,14 +378,9 @@ impl Beacon for DrandBeacon { async fn entry(&self, round: u64) -> anyhow::Result { if let Some(cached_entry) = self.verified_beacons.get(&round) { - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::CACHE) - .inc(); return Ok(Arc::unwrap_or_clone(cached_entry)); } - let _timer = metrics::DRAND_HTTP_FETCH_TIME.start_timer(); - async fn fetch_entry_from_url(url: impl reqwest::IntoUrl) -> anyhow::Result { let resp: BeaconEntryJson = global_http_client() .get(url) @@ -434,25 +423,16 @@ impl Beacon for DrandBeacon { humantime::format_duration(dur) ); }) - .await - .inspect_err(|_| { - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::HTTP_ERROR) - .inc(); - })?; + .await?; // Callers assume the entry is for the round they asked for. Round 0 is served // as "latest", so it answers with a different round by design: // - if round != 0 && entry.round() != round { - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::HTTP_ERROR) - .inc(); - anyhow::bail!("drand returned round {} for round {round}", entry.round()); - } + anyhow::ensure!( + round == 0 || entry.round() == round, + "drand returned round {} for round {round}", + entry.round() + ); self.cache_fetched_entry(&entry); - metrics::DRAND_ENTRY_SOURCE_TOTAL - .get_or_create(&metrics::HTTP) - .inc(); Ok(entry) } diff --git a/src/beacon/metrics.rs b/src/beacon/metrics.rs deleted file mode 100644 index 4e3f4a6cb449..000000000000 --- a/src/beacon/metrics.rs +++ /dev/null @@ -1,43 +0,0 @@ -// Copyright 2019-2026 ChainSafe Systems -// SPDX-License-Identifier: Apache-2.0, MIT - -use prometheus_client::{ - encoding::EncodeLabelSet, - metrics::{counter::Counter, family::Family, histogram::Histogram}, -}; -use std::sync::LazyLock; - -#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet, derive_more::Constructor)] -pub struct DrandSourceLabel { - pub source: &'static str, -} - -// drand_entry_source_total -pub const CACHE: DrandSourceLabel = DrandSourceLabel::new("cache"); -pub const HTTP: DrandSourceLabel = DrandSourceLabel::new("http"); -pub const HTTP_ERROR: DrandSourceLabel = DrandSourceLabel::new("http_error"); - -/// Counts every round served by [`crate::beacon::Beacon::entry`], labelled by where it -/// came from. -pub static DRAND_ENTRY_SOURCE_TOTAL: LazyLock> = - LazyLock::new(|| { - let metric = Family::default(); - crate::metrics::default_registry().register( - "drand_entry_source_total", - "Total number of drand rounds served, by source", - metric.clone(), - ); - metric - }); - -/// Wall-clock duration of a drand HTTP fetch, covering the whole retry chain across -/// every configured server rather than a single attempt. -pub static DRAND_HTTP_FETCH_TIME: LazyLock = LazyLock::new(|| { - let metric = crate::metrics::default_histogram(); - crate::metrics::default_registry().register( - "drand_http_fetch_time", - "Duration of a drand HTTP round fetch, including retries across servers", - metric.clone(), - ); - metric -}); diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index 2eb08507da9f..b4b536c4f5d4 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -4,7 +4,6 @@ pub mod beacon_entries; mod drand; mod drand_pb; -pub mod metrics; pub mod signatures; pub use beacon_entries::*; pub use drand::*; diff --git a/src/beacon/tests/drand.rs b/src/beacon/tests/drand.rs index f5a34e3658a8..b5611b30fbf8 100644 --- a/src/beacon/tests/drand.rs +++ b/src/beacon/tests/drand.rs @@ -308,36 +308,3 @@ async fn beacon_entries_for_block_covers_null_rounds_quicknet() { ); } } - -#[test] -#[serial_test::serial] -fn verified_beacons_cache_metrics_are_uniquely_named() { - use crate::networks::ChainConfig; - - crate::metrics::reset_collector_registry(); - // Mainnet has three drand points: Incentinet, Mainnet and Quicknet. - let schedule = ChainConfig::mainnet().get_beacon_schedule(1598306400); - assert_eq!(schedule.0.len(), 3); - - let mut encoded = String::new(); - prometheus_client::encoding::text::encode_registry( - &mut encoded, - &crate::metrics::collector_registry(), - ) - .unwrap(); - - let families: Vec<_> = encoded - .lines() - .filter_map(|line| line.strip_prefix("# HELP ")) - .filter(|line| line.starts_with("cache_verified_beacons")) - .map(|line| line.split_whitespace().next().unwrap_or_default()) - .collect(); - - // Five metrics (size/len/cap/hits/misses) for each of the three beacons. - assert_eq!(families.len(), 15); - assert_eq!( - families.iter().unique().count(), - families.len(), - "duplicate cache metric families: {families:?}" - ); -} From 377152b2a2289b448bc84813b87619ecffbb0ba2 Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Tue, 25 Aug 2026 14:13:37 -0400 Subject: [PATCH 05/12] chore: use block delay in secs --- src/chain_sync/chain_follower.rs | 12 ++++++++---- src/libp2p/service.rs | 10 ++++++++-- src/networks/mod.rs | 8 -------- 3 files changed, 16 insertions(+), 14 deletions(-) diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index ce02a0522648..350ed1353164 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -327,6 +327,10 @@ async fn chain_follower( ), Ok(true) ) { + info!( + round = entry.round(), + "verified drand entry from gossipsub" + ); last_drand_entry.store( Utc::now().timestamp().max(0) as u64, Ordering::Relaxed, @@ -520,12 +524,12 @@ async fn drand_gossip_watchdog( last_drand_entry: Arc, cancellation_token: CancellationToken, ) { - let chain_config = state_manager.chain_config(); - let Some(period_secs) = chain_config.drand_gossip_period_secs() else { + if state_manager.beacon_schedule().unchained_beacon().is_none() { return; - }; + } - let deadline = Duration::from_secs(period_secs.div_ceil(2)); + let deadline = + Duration::from_secs(u64::from(state_manager.chain_config().block_delay_secs).div_ceil(2)); let mut ticker = tokio::time::interval(deadline); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index 3ca584162eec..33495256816a 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -32,7 +32,7 @@ use libp2p::{ tcp, yamux, }; use nonzero_ext::nonzero; -use quick_protobuf::deserialize_from_slice; +use quick_protobuf::{BytesReader, MessageRead as _}; use tokio_stream::wrappers::IntervalStream; use tracing::{debug, error, info, trace, warn}; @@ -271,6 +271,7 @@ impl Libp2pService { .behaviour_mut() .subscribe(&topic) .with_context(|| format!("Failed to subscribe gossipsub topic {topic}"))?; + info!("Subscribed to gossipsub topic {topic} ({kind:?})"); pubsub_topic_kinds.insert(topic.hash(), kind); } @@ -739,8 +740,13 @@ async fn handle_gossip_event( } } Some(PubsubTopic::Drand) => { - match deserialize_from_slice::(&message) { + let mut reader = BytesReader::from_bytes(&message); + match PublicRandResponse::from_reader(&mut reader, &message) { Ok(r) => { + info!( + "Received drand round {} from peer {source:?} on {topic}", + r.round + ); emit_event( network_sender_out, NetworkEvent::PubsubMessage { diff --git a/src/networks/mod.rs b/src/networks/mod.rs index a8e9eaeace7a..58e906615cc8 100644 --- a/src/networks/mod.rs +++ b/src/networks/mod.rs @@ -503,14 +503,6 @@ impl ChainConfig { .collect() } - /// Round interval, in seconds, of the drand network Forest subscribes to over - /// gossipsub. `None` when the network has no unchained drand point. - pub fn drand_gossip_period_secs(&self) -> Option { - self.drand_points() - .find(|p| p.config.network_type.is_unchained()) - .map(|p| p.config.chain_info.period as u64) - } - pub fn get_beacon_schedule(&self, genesis_ts: u64) -> BeaconSchedule { BeaconSchedule( self.drand_points() From ac6700ef32cc36acc0e6ef2b7bb3989c4f5ac850 Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Tue, 25 Aug 2026 16:05:39 -0400 Subject: [PATCH 06/12] chore: introduce resubscription --- src/chain_sync/chain_follower.rs | 41 ++++++++++++++++++++++---------- src/libp2p/behaviour.rs | 22 +++++++++-------- src/libp2p/service.rs | 30 ++++++++++++++++++++++- 3 files changed, 70 insertions(+), 23 deletions(-) diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index 7eb40b1a833c..28bef4fefda8 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -18,23 +18,16 @@ use super::network_context::SyncNetworkContext; use crate::{ - beacon::{Beacon, BeaconEntry}, - blocks::{Block, FullTipset, Tipset, TipsetKey}, - chain::{ChainStore, index::ResolveNullTipset}, - chain_sync::{ + beacon::{Beacon, BeaconEntry}, blocks::{Block, FullTipset, Tipset, TipsetKey}, chain::{ChainStore, index::ResolveNullTipset}, chain_sync::{ ForkSyncInfo, ForkSyncStage, SyncStatus, SyncStatusReport, TipsetValidator, bad_block_cache::{BadBlockCache, SeenBlockCache}, metrics, tipset_syncer::{TipsetSyncerError, validate_tipset}, validation::GossipBlockValidator, + }, libp2p::{NetworkEvent, NetworkMessage, PubsubMessage, PubsubTopic, hello::HelloRequest}, message_pool::MessagePool, networks::calculate_expected_epoch, prelude::*, shim::clock::ChainEpoch, state_manager::StateManager, utils::{ + flume::FlumeSenderExt as _, + misc::env::env_or_default_logged, }, - libp2p::{NetworkEvent, PubsubMessage, hello::HelloRequest}, - message_pool::MessagePool, - networks::calculate_expected_epoch, - prelude::*, - shim::clock::ChainEpoch, - state_manager::StateManager, - utils::misc::env::env_or_default_logged, }; use arc_swap::ArcSwap; use chrono::Utc; @@ -357,10 +350,16 @@ async fn chain_follower( set.spawn({ let state_manager = state_manager.shallow_clone(); + let network = network.shallow_clone(); let last_drand_entry = last_drand_entry.clone(); let cancellation_token = cancellation_token.clone(); async move { - drand_gossip_watchdog(state_manager, last_drand_entry, cancellation_token).await; + drand_gossip_watchdog( + state_manager, + network, + last_drand_entry, + cancellation_token, + ).await; } }); @@ -526,6 +525,7 @@ async fn chain_follower( /// that epoch and fallback to fetch the beacon through HTTP async fn drand_gossip_watchdog( state_manager: StateManager, + network: SyncNetworkContext, last_drand_entry: Arc, cancellation_token: CancellationToken, ) { @@ -538,7 +538,11 @@ async fn drand_gossip_watchdog( let mut ticker = tokio::time::interval(deadline); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + + const MAX_CONSECUTIVE_MISSES: u32 = 3; + let mut stale = false; + let mut consecutive_misses = 0_u32; while cancellation_token .run_until_cancelled(ticker.tick()) @@ -548,6 +552,7 @@ async fn drand_gossip_watchdog( let last_seen = last_drand_entry.load(Ordering::Relaxed); let now = Utc::now().timestamp().max(0) as u64; if last_seen != 0 && now.saturating_sub(last_seen) < deadline.as_secs() { + consecutive_misses = 0; if stale { stale = false; info!("drand gossipsub entries are flowing again"); @@ -578,6 +583,18 @@ async fn drand_gossip_watchdog( if let Err(e) = beacon.entry(round).await { debug!("drand HTTP fallback for round {round} failed: {e:#}"); } + + consecutive_misses += 1; + if consecutive_misses >= MAX_CONSECUTIVE_MISSES { + consecutive_misses = 0; + warn!( + misses = MAX_CONSECUTIVE_MISSES, + "forcing a drand topic re-subscription", + ); + network + .network_send() + .send_or_warn(NetworkMessage::ResubscribeTopic(PubsubTopic::Drand)); + } } } diff --git a/src/libp2p/behaviour.rs b/src/libp2p/behaviour.rs index 37cf9df89af6..027f50b839fb 100644 --- a/src/libp2p/behaviour.rs +++ b/src/libp2p/behaviour.rs @@ -24,16 +24,9 @@ use crate::{ }; use ahash::{HashMap, HashSet}; use libp2p::{ - Multiaddr, allow_block_list, connection_limits, - gossipsub::{ - self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAuthenticity, MessageId, - PublishError, SubscriptionError, ValidationMode, WhitelistSubscriptionFilter, - }, - identity::{Keypair, PeerId}, - kad::QueryId, - metrics::{Metrics, Recorder}, - ping, request_response, - swarm::NetworkBehaviour, + Multiaddr, allow_block_list, connection_limits, gossipsub::{ + self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAuthenticity, MessageId, PublishError, SubscriptionError, TopicHash, ValidationMode, WhitelistSubscriptionFilter, + }, identity::{Keypair, PeerId}, kad::QueryId, metrics::{Metrics, Recorder}, ping, request_response, swarm::NetworkBehaviour, }; use tracing::info; @@ -230,6 +223,15 @@ impl ForestBehaviour { self.gossipsub.subscribe(topic) } + /// Unsubscribe from a gossip topic. + pub fn unsubscribe(&mut self, topic: &Topic) -> bool { + self.gossipsub.unsubscribe(topic) + } + + pub fn mesh_peers(&self, topic_hash: &TopicHash) -> impl Iterator { + self.gossipsub.mesh_peers(topic_hash) + } + /// Returns a set of peer ids pub fn peers(&self) -> &HashSet { self.discovery.peers() diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index 33495256816a..dbd4fbd0d1b4 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -173,6 +173,7 @@ pub enum NetworkMessage { topic: IdentTopic, message: Vec, }, + ResubscribeTopic(PubsubTopic), ChainExchangeRequest { peer_id: PeerId, request: ChainExchangeRequest, @@ -400,7 +401,9 @@ impl Libp2pService { bitswap_request_manager.shallow_clone(), message, &self.network_sender_out, - &self.peer_manager).await; + &self.peer_manager, + &pubsub_topic_kinds, + ).await; } None => { break; } }, @@ -503,6 +506,7 @@ async fn handle_network_message( message: NetworkMessage, network_sender_out: &Sender, peer_manager: &PeerManager, + pubsub_topic_kinds: &HashMap, ) { match message { NetworkMessage::PubsubMessage { topic, message } => { @@ -518,6 +522,30 @@ async fn handle_network_message( } } } + NetworkMessage::ResubscribeTopic(pubsub_topic) => { + for (topic_hash, kind) in pubsub_topic_kinds.iter() { + if !matches!(kind, pubsub_topic) { + continue; + } + + let topic = IdentTopic::new(topic_hash.as_str()); + let mesh_peers_before = swarm + .behaviour() + .mesh_peers(&topic_hash) + .count(); + + swarm.behaviour_mut().unsubscribe(&topic); + + match swarm.behaviour_mut().subscribe(&topic) { + Ok(_) => info!( + %topic, + mesh_peers_before, + "re-subscribed to drand topic after repeated silence" + ), + Err(e) => warn!(%topic, "failed to re-subscribe to drand topic: {e}"), + } + } + } NetworkMessage::HelloRequest { peer_id, request, From 0f17015a4b1fb65c3452b4bcc059ad6fdd879e72 Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Wed, 26 Aug 2026 00:16:32 -0400 Subject: [PATCH 07/12] chore: add tests --- src/beacon/mod.rs | 1 + src/beacon/signatures/mod.rs | 2 +- src/beacon/tests/fake_drand.rs | 92 ++++++++++++ src/chain_sync/chain_follower.rs | 23 +-- src/libp2p/behaviour.rs | 13 +- src/libp2p/mod.rs | 1 + src/libp2p/service.rs | 9 +- src/libp2p/tests/drand_gossip_tests.rs | 164 ++++++++++++++++++++++ src/libp2p/tests/gossipsub_filter_test.rs | 8 +- 9 files changed, 289 insertions(+), 24 deletions(-) create mode 100644 src/beacon/tests/fake_drand.rs create mode 100644 src/libp2p/tests/drand_gossip_tests.rs diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index b4b536c4f5d4..1a43cd193e2a 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -16,4 +16,5 @@ pub mod tests { // `pub` so that helpers such as `drand::new_beacon_quicknet` can be shared with // tests in other modules. pub mod drand; + pub mod fake_drand; } diff --git a/src/beacon/signatures/mod.rs b/src/beacon/signatures/mod.rs index 19cb9b90d456..98613685f807 100644 --- a/src/beacon/signatures/mod.rs +++ b/src/beacon/signatures/mod.rs @@ -13,7 +13,7 @@ use rayon::prelude::*; pub use bls_signatures::{PublicKey as PublicKeyOnG1, Signature as SignatureOnG2}; // See -const CSUITE_G1: &[u8] = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_"; +pub const CSUITE_G1: &[u8] = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_"; const CSUITE_G2: &[u8] = b"BLS_SIG_BLS12381G2_XMD:SHA-256_SSWU_RO_NUL_"; #[derive(Debug, Clone, Eq, PartialEq, derive_more::Deref)] diff --git a/src/beacon/tests/fake_drand.rs b/src/beacon/tests/fake_drand.rs new file mode 100644 index 000000000000..d0ac1f870297 --- /dev/null +++ b/src/beacon/tests/fake_drand.rs @@ -0,0 +1,92 @@ +use crate::beacon::{Beacon, BeaconEntry, ChainInfo, DrandBeacon, DrandConfig, DrandNetwork}; +use blstrs::{G1Projective, G2Projective, Scalar}; +use group::{Curve, Group}; + +pub const FAKE_DRAND_GENESIS_TIME: i32 = 1_692_803_367; +pub const FAKE_DRAND_PERIOD: i32 = 3; + +pub const TEST_FIL_GENESIS_TIME: u64 = 1_598_306_400; +pub const TEST_FIL_BLOCK_DELAY: u64 = 30; + +pub struct FakeDrand { + secret: Scalar, + config: DrandConfig<'static>, +} + +impl FakeDrand { + pub fn new(servers: Vec, period: i32, genesis_time: i32) -> Self { + let secret = Scalar::from(0xC0FFEEu64); + let public = G2Projective::generator() * secret; + let public_key = hex::encode(public.to_affine().to_compressed()); + Self { + secret, + config: DrandConfig { + servers, + chain_info: ChainInfo { + public_key: public_key.into(), + period, + genesis_time, + hash: "0011".repeat(16).into(), + group_hash: "00".repeat(32).into(), + }, + network_type: DrandNetwork::Quicknet, // unchained + }, + } + } + + // sing H(round) on G1, exactly what `verifies_entries` checks for unchained + pub fn entry(&self, round: u64) -> BeaconEntry { + let msg = BeaconEntry::message_unchained(round); + let point = + G1Projective::hash_to_curve(msg.as_ref(), crate::beacon::signatures::CSUITE_G1, &[]); + let point = point * self.secret; + BeaconEntry::new(round, point.to_affine().to_compressed().to_vec()) + } + + // encode PublicRandResponse to protobuf + pub fn to_protobuf(&self, round: u64) -> Vec { + let entry = self.entry(round); + let mut out = Vec::new(); + let mut w = quick_protobuf::Writer::new(&mut out); + quick_protobuf::MessageWrite::write_message( + &crate::beacon::drand_pb::PublicRandResponse { + round, + signature: entry.signature().to_vec(), + }, + &mut w, + ) + .unwrap(); + out + } + + pub fn to_json(&self, round: u64) -> serde_json::Value { + let entry = self.entry(round); + serde_json::json!({ + "round": round, + "randomness": "00".repeat(32), + "signature": hex::encode(entry.signature()), + "previous_signature": null, + }) + } + + pub fn beacon(&self, genesis_ts: u64, block_delay: u64) -> DrandBeacon { + DrandBeacon::new(genesis_ts, block_delay, &self.config) + } + + pub fn chain_info_hash(&self) -> String { + self.config.chain_info.hash.to_string() + } +} + +// just test the secret and public keys are correctly validating +#[test] +fn fake_drand_entries_verify() { + let d = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let beacon = d.beacon(TEST_FIL_GENESIS_TIME, TEST_FIL_BLOCK_DELAY); + let entries: Vec<_> = (1..=5).map(|r| d.entry(r)).collect(); + assert!( + beacon + .verify_entries(&entries, &BeaconEntry::default()) + .unwrap() + ); +} diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index 28bef4fefda8..9e35c9a9d4cc 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -18,16 +18,23 @@ use super::network_context::SyncNetworkContext; use crate::{ - beacon::{Beacon, BeaconEntry}, blocks::{Block, FullTipset, Tipset, TipsetKey}, chain::{ChainStore, index::ResolveNullTipset}, chain_sync::{ + beacon::{Beacon, BeaconEntry}, + blocks::{Block, FullTipset, Tipset, TipsetKey}, + chain::{ChainStore, index::ResolveNullTipset}, + chain_sync::{ ForkSyncInfo, ForkSyncStage, SyncStatus, SyncStatusReport, TipsetValidator, bad_block_cache::{BadBlockCache, SeenBlockCache}, metrics, tipset_syncer::{TipsetSyncerError, validate_tipset}, validation::GossipBlockValidator, - }, libp2p::{NetworkEvent, NetworkMessage, PubsubMessage, PubsubTopic, hello::HelloRequest}, message_pool::MessagePool, networks::calculate_expected_epoch, prelude::*, shim::clock::ChainEpoch, state_manager::StateManager, utils::{ - flume::FlumeSenderExt as _, - misc::env::env_or_default_logged, }, + libp2p::{NetworkEvent, NetworkMessage, PubsubMessage, PubsubTopic, hello::HelloRequest}, + message_pool::MessagePool, + networks::calculate_expected_epoch, + prelude::*, + shim::clock::ChainEpoch, + state_manager::StateManager, + utils::{flume::FlumeSenderExt as _, misc::env::env_or_default_logged}, }; use arc_swap::ArcSwap; use chrono::Utc; @@ -354,12 +361,8 @@ async fn chain_follower( let last_drand_entry = last_drand_entry.clone(); let cancellation_token = cancellation_token.clone(); async move { - drand_gossip_watchdog( - state_manager, - network, - last_drand_entry, - cancellation_token, - ).await; + drand_gossip_watchdog(state_manager, network, last_drand_entry, cancellation_token) + .await; } }); diff --git a/src/libp2p/behaviour.rs b/src/libp2p/behaviour.rs index 027f50b839fb..8a2fd234444d 100644 --- a/src/libp2p/behaviour.rs +++ b/src/libp2p/behaviour.rs @@ -24,9 +24,16 @@ use crate::{ }; use ahash::{HashMap, HashSet}; use libp2p::{ - Multiaddr, allow_block_list, connection_limits, gossipsub::{ - self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAuthenticity, MessageId, PublishError, SubscriptionError, TopicHash, ValidationMode, WhitelistSubscriptionFilter, - }, identity::{Keypair, PeerId}, kad::QueryId, metrics::{Metrics, Recorder}, ping, request_response, swarm::NetworkBehaviour, + Multiaddr, allow_block_list, connection_limits, + gossipsub::{ + self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAuthenticity, MessageId, + PublishError, SubscriptionError, TopicHash, ValidationMode, WhitelistSubscriptionFilter, + }, + identity::{Keypair, PeerId}, + kad::QueryId, + metrics::{Metrics, Recorder}, + ping, request_response, + swarm::NetworkBehaviour, }; use tracing::info; diff --git a/src/libp2p/mod.rs b/src/libp2p/mod.rs index f9002fbd16c3..d7eb355c289e 100644 --- a/src/libp2p/mod.rs +++ b/src/libp2p/mod.rs @@ -25,5 +25,6 @@ pub use self::{config::*, peer_manager::*, service::*}; #[cfg(test)] mod tests { mod decode_test; + mod drand_gossip_tests; mod gossipsub_filter_test; } diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index dbd4fbd0d1b4..49c05ccced54 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -89,7 +89,7 @@ pub const PUBSUB_DRAND_STR: &str = "/drand/pubsub/v0.0.0"; /// Gossipsub topics Forest uses. Subscription, the subscription-filter /// whitelist, and peer-score params all iterate the variants, so adding one is /// handled everywhere. -#[derive(Copy, Clone, Debug, strum::EnumIter, derive_more::Display)] +#[derive(Copy, Clone, Debug, strum::EnumIter, derive_more::Display, Eq, PartialEq)] pub enum PubsubTopic { #[display("{PUBSUB_BLOCK_STR}")] Blocks, @@ -524,15 +524,12 @@ async fn handle_network_message( } NetworkMessage::ResubscribeTopic(pubsub_topic) => { for (topic_hash, kind) in pubsub_topic_kinds.iter() { - if !matches!(kind, pubsub_topic) { + if !pubsub_topic.eq(kind) { continue; } let topic = IdentTopic::new(topic_hash.as_str()); - let mesh_peers_before = swarm - .behaviour() - .mesh_peers(&topic_hash) - .count(); + let mesh_peers_before = swarm.behaviour().mesh_peers(&topic_hash).count(); swarm.behaviour_mut().unsubscribe(&topic); diff --git a/src/libp2p/tests/drand_gossip_tests.rs b/src/libp2p/tests/drand_gossip_tests.rs new file mode 100644 index 000000000000..2e78c2db604b --- /dev/null +++ b/src/libp2p/tests/drand_gossip_tests.rs @@ -0,0 +1,164 @@ +use std::{sync::Arc, time::Duration}; + +use futures::StreamExt as _; +use libp2p::{ + Swarm, + gossipsub::{self, IdentTopic}, + swarm::SwarmEvent, +}; +use libp2p_swarm_test::SwarmExt as _; +use quick_protobuf::{BytesReader, MessageRead}; + +use crate::libp2p::{PUBSUB_DRAND_STR, build_gossipsub}; +use crate::networks::GenesisNetworkName; +use crate::{ + beacon::{ + Beacon, BeaconEntry, PublicRandResponse, + tests::fake_drand::{ + FAKE_DRAND_GENESIS_TIME, FAKE_DRAND_PERIOD, FakeDrand, TEST_FIL_BLOCK_DELAY, + TEST_FIL_GENESIS_TIME, + }, + }, + libp2p::{Gossipsub, PubsubTopicCfg}, +}; + +#[tokio::test] +async fn gossip_rounds_are_verified_and_cached() { + let drand = FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + + let beacon = drand.beacon(TEST_FIL_GENESIS_TIME, TEST_FIL_BLOCK_DELAY); + let hash = drand.chain_info_hash(); + + let topic = IdentTopic::new(format!("{PUBSUB_DRAND_STR}/{hash}")); + + // `PubsubTopicCfg` borrows, so these have to outlive the swarm construction. + // The whitelist must carry the *fake* chain hash, otherwise the node refuses + // to subscribe to the topic the relay publishes on. + let network_name: GenesisNetworkName = "testdrandgossipsub".into(); + let drand_chain_hashes = vec![hash]; + let cfg = PubsubTopicCfg { + network_name: &network_name, + drand_chain_hashes: &drand_chain_hashes, + }; + + let mut node = Swarm::new_ephemeral_tokio(|id| build_gossipsub(&id, cfg).unwrap()); + + let mut relay = Swarm::new_ephemeral_tokio(|id| { + gossipsub::Behaviour::new( + gossipsub::MessageAuthenticity::Signed(id), + gossipsub::ConfigBuilder::default().build().unwrap(), + ) + .unwrap() + }); + + node.listen().with_memory_addr_external().await; + relay.connect(&mut node).await; + + relay.behaviour_mut().subscribe(&topic).unwrap(); + node.behaviour_mut().subscribe(&topic).unwrap(); + + wait_until_meshed(&mut node, &mut relay, &topic).await; + + let mut received = Vec::new(); + for round in 1..=5u64 { + relay + .behaviour_mut() + .publish(topic.clone(), drand.to_protobuf(round)) + .unwrap(); + let data = tokio::time::timeout(Duration::from_secs(5), async { + loop { + tokio::select! { + _ = relay.select_next_some() => {}, + ev = node.select_next_some() => { + if let SwarmEvent::Behaviour(gossipsub::Event::Message { message, .. }) = ev { + break message.data; + } + } + } + } + }).await.expect("no gossip message"); + + let mut reader = BytesReader::from_bytes(&data); + let decoded = PublicRandResponse::from_reader(&mut reader, &data).unwrap(); + received.push(BeaconEntry::new(decoded.round, decoded.signature)); + } + + assert_eq!(received.len(), 5); + assert!( + beacon + .verify_entries(&received, &BeaconEntry::default()) + .unwrap() + ); + + // verify every round is now served from cache. + for round in 1..=5u64 { + assert_eq!(beacon.entry(round).await.unwrap().round(), round); + } +} + +async fn wait_until_meshed( + node: &mut Swarm, + relay: &mut Swarm, + topic: &IdentTopic, +) { + let hash = topic.hash(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + if relay.behaviour().mesh_peers(&hash).next().is_some() + && node.behaviour().mesh_peers(&hash).next().is_some() + { + return; + } + + tokio::select! { + _ = node.select_next_some() => {} + _ = relay.select_next_some() => {} + } + } + }) + .await + .expect("drand topic mesh never formed"); +} + +#[tokio::test] +async fn silence_past_deadline_fallback_to_http() { + use axum::{routing::get, Router, extract::Path, Json}; + use std::sync::atomic::{AtomicUsize, Ordering}; + + // mocks drand HTTP + let hits = Arc::new(AtomicUsize::new(0)); + let signer = Arc::new(FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME)); + + let app = { + let (hits, signer) = (hits.clone(), signer.clone()); + Router::new().route( + "/{hash}/public/{round}", + get(move |Path((_hash, round)): Path<(String, u64)>| { + let (hits, signer) = (hits.clone(), signer.clone()); + async move { + hits.fetch_add(1, Ordering::Relaxed); + Json(signer.to_json(round)) + } + }) + ) + }; + + let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); + let base: url::Url = format!("http://{}/", listener.local_addr().unwrap()).parse().unwrap(); + tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); + + let drand = FakeDrand::new(vec![base], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); + let beacon = drand.beacon(TEST_FIL_GENESIS_TIME, TEST_FIL_BLOCK_DELAY); + + // sanity check + assert_eq!(hits.load(Ordering::Relaxed), 0); + + // first fetch + let fetched = beacon.entry(42).await.unwrap(); + assert_eq!(fetched.round(), 42); + assert_eq!(hits.load(Ordering::Relaxed), 1, "expected one HTTP fetch"); + + // second call, same round, should fetch from cache + beacon.entry(42).await.unwrap(); + assert_eq!(hits.load(Ordering::Relaxed), 1, "second call must not reach HTTP"); +} \ No newline at end of file diff --git a/src/libp2p/tests/gossipsub_filter_test.rs b/src/libp2p/tests/gossipsub_filter_test.rs index 81e6e6a5830c..774bbdde6480 100644 --- a/src/libp2p/tests/gossipsub_filter_test.rs +++ b/src/libp2p/tests/gossipsub_filter_test.rs @@ -24,20 +24,20 @@ const NETWORK: &str = "testnetname"; const DRAND_HASH: &str = "52db9ba70e0cc0f6eaf7803dd07447a1f5477735fd3f661792ba94600c84e971"; /// Owns what [`PubsubTopicCfg`] borrows. -struct TopicCfgOwner { +pub(in crate::libp2p) struct TopicCfgOwner { network_name: GenesisNetworkName, drand_chain_hashes: Vec, } impl TopicCfgOwner { - fn new() -> Self { + pub(in crate::libp2p) fn new() -> Self { Self { network_name: NETWORK.into(), drand_chain_hashes: vec![DRAND_HASH.to_string()], } } - fn cfg(&self) -> PubsubTopicCfg<'_> { + pub(in crate::libp2p) fn cfg(&self) -> PubsubTopicCfg<'_> { PubsubTopicCfg { network_name: &self.network_name, drand_chain_hashes: &self.drand_chain_hashes, @@ -56,8 +56,8 @@ fn allowed_topics() -> Vec { /// Swarm using Forest's subscription filter (the code under test). fn filtered_swarm() -> Swarm { + let owner = TopicCfgOwner::new(); Swarm::new_ephemeral_tokio(|identity| { - let owner = TopicCfgOwner::new(); build_gossipsub(&identity, owner.cfg()).expect("failed to build gossipsub") }) } From f146c60572a99ed5d4c3110147d2ba64c942a043 Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Wed, 26 Aug 2026 00:25:31 -0400 Subject: [PATCH 08/12] chore: update changelog --- CHANGELOG.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/CHANGELOG.md b/CHANGELOG.md index baa06e8d0d38..a7776c1e5d93 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,8 @@ ### Added +- [#7414](https://github.com/ChainSafe/forest/issues/7414): Forest now subscribes to the drand gossipsub topic, verifying and caching the beacon entries it receives there. The drand HTTP endpoints are only queried when no entry has arrived for half a chain epoch. + ### Changed - [#7535](https://github.com/ChainSafe/forest/pull/7535): `Filecoin.Version` now reports the API version of the endpoint being served (`1.5.0` over `/rpc/v0`, `2.3.0` over `/rpc/v1`), matching Lotus, and `Filecoin.SyncSubmitBlock` no longer requires the node to be in the `Synced` state and waits up to one block time for the submitted block to become the chain head. Together these let Forest act as the full node for an external block producer such as `lotus-miner` or `curio` (verified on a local devnet, calibnet/mainnet tests to come). From 3d19243b33bd19b905564d2908a47915db191062 Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Wed, 26 Aug 2026 06:31:31 -0400 Subject: [PATCH 09/12] chore: fix lint --- src/chain_sync/chain_follower.rs | 4 ++-- src/libp2p/service.rs | 6 +++--- src/libp2p/tests/drand_gossip_tests.rs | 22 ++++++++++++++++------ 3 files changed, 21 insertions(+), 11 deletions(-) diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index 9e35c9a9d4cc..aaf4df9a25c9 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -523,9 +523,9 @@ async fn chain_follower( Ok(()) } -/// Watch the drand gossipsub for staleness, if a drand breacon entry +/// Watch the `drand` `gossipsub` topic for staleness: if a `drand` beacon entry /// is not received in half a chain epoch then we consider it stale for -/// that epoch and fallback to fetch the beacon through HTTP +/// that epoch and fall back to fetching the beacon over HTTP. async fn drand_gossip_watchdog( state_manager: StateManager, network: SyncNetworkContext, diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index 49c05ccced54..2f5d16c0964a 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -83,7 +83,7 @@ crate::def_is_env_truthy!(libp2p_metrics_enabled, "FOREST_LIBP2P_METRICS_ENABLED pub const PUBSUB_BLOCK_STR: &str = "/fil/blocks"; /// `Gossipsub` Filecoin messages topic identifier. pub const PUBSUB_MSG_STR: &str = "/fil/msgs"; -/// `Gossipsub` drand randomness topic identifier. +/// `Gossipsub` `drand` randomness topic identifier. pub const PUBSUB_DRAND_STR: &str = "/drand/pubsub/v0.0.0"; /// Gossipsub topics Forest uses. Subscription, the subscription-filter @@ -162,7 +162,7 @@ pub enum PubsubMessage { Block(GossipBlock), /// Messages that come over the message topic Message(SignedMessage), - /// Messages that come over the drand topic + /// Messages that come over the `drand` topic DrandEntry(BeaconEntry), } @@ -529,7 +529,7 @@ async fn handle_network_message( } let topic = IdentTopic::new(topic_hash.as_str()); - let mesh_peers_before = swarm.behaviour().mesh_peers(&topic_hash).count(); + let mesh_peers_before = swarm.behaviour().mesh_peers(topic_hash).count(); swarm.behaviour_mut().unsubscribe(&topic); diff --git a/src/libp2p/tests/drand_gossip_tests.rs b/src/libp2p/tests/drand_gossip_tests.rs index 2e78c2db604b..58a3872a2814 100644 --- a/src/libp2p/tests/drand_gossip_tests.rs +++ b/src/libp2p/tests/drand_gossip_tests.rs @@ -122,12 +122,16 @@ async fn wait_until_meshed( #[tokio::test] async fn silence_past_deadline_fallback_to_http() { - use axum::{routing::get, Router, extract::Path, Json}; + use axum::{Json, Router, extract::Path, routing::get}; use std::sync::atomic::{AtomicUsize, Ordering}; // mocks drand HTTP let hits = Arc::new(AtomicUsize::new(0)); - let signer = Arc::new(FakeDrand::new(vec![], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME)); + let signer = Arc::new(FakeDrand::new( + vec![], + FAKE_DRAND_PERIOD, + FAKE_DRAND_GENESIS_TIME, + )); let app = { let (hits, signer) = (hits.clone(), signer.clone()); @@ -139,12 +143,14 @@ async fn silence_past_deadline_fallback_to_http() { hits.fetch_add(1, Ordering::Relaxed); Json(signer.to_json(round)) } - }) + }), ) }; let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); - let base: url::Url = format!("http://{}/", listener.local_addr().unwrap()).parse().unwrap(); + let base: url::Url = format!("http://{}/", listener.local_addr().unwrap()) + .parse() + .unwrap(); tokio::spawn(async move { axum::serve(listener, app).await.unwrap() }); let drand = FakeDrand::new(vec![base], FAKE_DRAND_PERIOD, FAKE_DRAND_GENESIS_TIME); @@ -160,5 +166,9 @@ async fn silence_past_deadline_fallback_to_http() { // second call, same round, should fetch from cache beacon.entry(42).await.unwrap(); - assert_eq!(hits.load(Ordering::Relaxed), 1, "second call must not reach HTTP"); -} \ No newline at end of file + assert_eq!( + hits.load(Ordering::Relaxed), + 1, + "second call must not reach HTTP" + ); +} From 2f8061afd18a604c2f32aaf17ab2532ba631c42a Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Mon, 31 Aug 2026 09:10:44 -0400 Subject: [PATCH 10/12] chore: remove changelog entry, place proto permalink, scoped const, include drand verify limiter --- CHANGELOG.md | 2 -- docs/docs/users/reference/env_variables.md | 1 + proto/drand_pb.proto | 1 + src/beacon/signatures/mod.rs | 2 +- src/beacon/tests/fake_drand.rs | 2 +- src/chain_sync/chain_follower.rs | 23 ++++++++++++++++++++++ 6 files changed, 27 insertions(+), 4 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0fe2626be138..3f38594c7b5f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -32,8 +32,6 @@ - [#7555](https://github.com/ChainSafe/forest/issues/7555): Added `--as-default` to `forest-wallet import`. - [#7414](https://github.com/ChainSafe/forest/issues/7414): New `drand_http_fetch_total` metric, counting the drand rounds that had to be fetched over HTTP rather than served from the in-memory cache. -- [#7414](https://github.com/ChainSafe/forest/issues/7414): Forest now subscribes to the drand gossipsub topic, verifying and caching the beacon entries it receives there. The drand HTTP endpoints are only queried when no entry has arrived for half a chain epoch. - ### Changed - [#7535](https://github.com/ChainSafe/forest/pull/7535): `Filecoin.Version` now reports the API version of the endpoint being served (`1.5.0` over `/rpc/v0`, `2.3.0` over `/rpc/v1`), matching Lotus, and `Filecoin.SyncSubmitBlock` no longer requires the node to be in the `Synced` state and waits up to one block time for the submitted block to become the chain head. Together these let Forest act as the full node for an external block producer such as `lotus-miner` or `curio` (verified on a local devnet, calibnet/mainnet tests to come). diff --git a/docs/docs/users/reference/env_variables.md b/docs/docs/users/reference/env_variables.md index 789bbb3708c5..fa8b322e8dd2 100644 --- a/docs/docs/users/reference/env_variables.md +++ b/docs/docs/users/reference/env_variables.md @@ -75,6 +75,7 @@ process. | `FOREST_MAX_CONCURRENT_INBOUND_CHAIN_EXCHANGE_REQUESTS` | positive integer | 32 | 32 | Maximum number of inbound chain exchange requests Forest will service concurrently. Excess requests are rejected with a `GoAway` response | | `FOREST_MAX_CONCURRENT_INBOUND_CHAIN_EXCHANGE_REQUESTS_PER_PEER` | positive integer | 4 | 4 | Per-peer cap on concurrent inbound chain exchange requests. Excess requests from a single peer are rejected with a `GoAway` response | | `FOREST_MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES` | positive integer | 16 | 16 | Bounds tipset fetches triggered by inbound `hello` requests that run concurrently; each chain-exchanges the peer's claimed head. Excess triggers are dropped, not queued. | +| `FOREST_MAX_CONCURRENT_DRAND_VERIFICATIONS` | positive integer | 4 | 4 | Bounds drand beacon entries from gossipsub verified concurrently; each costs a BLS pairing. Excess entries are dropped, not queued. | | `FOREST_MAX_OUTBOUND_CHAIN_EXCHANGE_RESPONSE_BYTES` | positive integer (bytes) | 10485760 (10 MiB) | 10485760 | Cap on the encoded byte size of a chain exchange response Forest serves to peers. Building stops as soon as the running encoded size would exceed this cap and the response is returned with `PartialResponse` status | | `FOREST_ETH_RPC_COMPUTE_STATE_ON_INDEX_MISS` | 1 or true | false | 1 | Allows Ethereum RPC methods to compute state trees on index miss | | `FOREST_ETH_RPC_COMPUTE_BLOOM_ON_MISS` | 1 or true | false | 1 | Allows `eth` block RPC methods to compute (and store) the block `logsBloom` when it is not already stored, otherwise such blocks report an all-ones bloom | diff --git a/proto/drand_pb.proto b/proto/drand_pb.proto index db63e3120baa..74319bd75a4b 100644 --- a/proto/drand_pb.proto +++ b/proto/drand_pb.proto @@ -2,6 +2,7 @@ syntax = "proto3"; package drand_pb; +// https://github.com/drand/drand/blob/v2.1.7/protobuf/drand/api.proto#L42-L53 message PublicRandResponse { uint64 round = 1; bytes signature = 2; diff --git a/src/beacon/signatures/mod.rs b/src/beacon/signatures/mod.rs index 98613685f807..bd297cc86b96 100644 --- a/src/beacon/signatures/mod.rs +++ b/src/beacon/signatures/mod.rs @@ -13,7 +13,7 @@ use rayon::prelude::*; pub use bls_signatures::{PublicKey as PublicKeyOnG1, Signature as SignatureOnG2}; // See -pub const CSUITE_G1: &[u8] = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_"; +pub(crate) const CSUITE_G1: &[u8] = b"BLS_SIG_BLS12381G1_XMD:SHA-256_SSWU_RO_NUL_"; const CSUITE_G2: &[u8] = b"BLS_SIG_BLS12381G2_XMD:SHA-256_SSWU_RO_NUL_"; #[derive(Debug, Clone, Eq, PartialEq, derive_more::Deref)] diff --git a/src/beacon/tests/fake_drand.rs b/src/beacon/tests/fake_drand.rs index 8cee7f538a21..8240fc0f365d 100644 --- a/src/beacon/tests/fake_drand.rs +++ b/src/beacon/tests/fake_drand.rs @@ -37,7 +37,7 @@ impl FakeDrand { } } - // sing H(round) on G1, exactly what `verifies_entries` checks for unchained + // sign H(round) on G1, exactly what `verify_entries` checks for unchained. pub fn entry(&self, round: u64) -> BeaconEntry { let msg = BeaconEntry::message_unchained(round); let point = diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index aaf4df9a25c9..66448601f3fe 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -230,6 +230,8 @@ async fn chain_follower( let hello_fetch_limiter = Arc::new(Semaphore::new(*MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES)); + let drand_verify_limiter = Arc::new(Semaphore::new(*MAX_CONCURRENT_DRAND_VERIFICATIONS)); + let last_drand_entry = Arc::new(AtomicU64::new(0)); let mut set = JoinSet::new(); @@ -248,6 +250,7 @@ async fn chain_follower( let hello_fetch_limiter = hello_fetch_limiter.shallow_clone(); let tipset_sender = tipset_sender.clone(); let last_drand_entry = last_drand_entry.clone(); + let drand_verify_limiter = drand_verify_limiter.shallow_clone(); async move { while let Ok(event) = network_rx.recv_async().await { inc_gossipsub_event_metrics(&event); @@ -318,9 +321,19 @@ async fn chain_follower( if entry.round() == 0 || entry.signature().is_empty() { continue; } + let Ok(permit) = + drand_verify_limiter.shallow_clone().try_acquire_owned() + else { + debug!( + round = entry.round(), + "dropping drand entry: too many verifications in flight" + ); + continue; + }; let beacon_schedule = state_manager.beacon_schedule().clone(); let last_drand_entry = last_drand_entry.clone(); tokio::task::spawn_blocking(move || { + let _permit = permit; let Some(beacon) = beacon_schedule.unchained_beacon() else { return; }; @@ -723,6 +736,16 @@ static MAX_CONCURRENT_HELLO_TRIGGERED_FETCHES: LazyLock = LazyLock::new(| .min(Semaphore::MAX_PERMITS) }); +/// Concurrency cap for `drand` entries verified from `gossipsub`. Excess is dropped, not queued. +static MAX_CONCURRENT_DRAND_VERIFICATIONS: LazyLock = LazyLock::new(|| { + env_or_default_logged( + "FOREST_MAX_CONCURRENT_DRAND_VERIFICATIONS", + nonzero!(4_usize), + ) + .get() + .min(Semaphore::MAX_PERMITS) +}); + /// Fetches a tipset off the event loop, forwarding a success into `tipset_sender` /// (the same channel miner tipsets use). Any `permit` is held for the fetch. fn spawn_tipset_fetch( From 426f2b027454ea2151af368395b38652185b4fdc Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Mon, 31 Aug 2026 14:07:39 -0400 Subject: [PATCH 11/12] chore: run `beacon.entry(...)` under cancellation_token.run_until_cancelled --- src/chain_sync/chain_follower.rs | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index 66448601f3fe..d2f2bd62211a 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -596,8 +596,15 @@ async fn drand_gossip_watchdog( continue; } }; - if let Err(e) = beacon.entry(round).await { - debug!("drand HTTP fallback for round {round} failed: {e:#}"); + // Inside the cancellation scope: `entry` retries with a 15s timeout across every + // configured server, so an in-flight fetch would otherwise hold up `join_all`. + match cancellation_token + .run_until_cancelled(beacon.entry(round)) + .await + { + None => return, + Some(Err(e)) => debug!("drand HTTP fallback for round {round} failed: {e:#}"), + Some(Ok(_)) => {} } consecutive_misses += 1; From ab1030f4f9fae57ed98142a098dde8b076adbe3a Mon Sep 17 00:00:00 2001 From: EclesioMeloJunior Date: Mon, 31 Aug 2026 14:37:57 -0400 Subject: [PATCH 12/12] chore: start the watchdog ticker correctly --- src/chain_sync/chain_follower.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/chain_sync/chain_follower.rs b/src/chain_sync/chain_follower.rs index d2f2bd62211a..f90a22f9a071 100644 --- a/src/chain_sync/chain_follower.rs +++ b/src/chain_sync/chain_follower.rs @@ -552,7 +552,7 @@ async fn drand_gossip_watchdog( let deadline = Duration::from_secs(u64::from(state_manager.chain_config().block_delay_secs).div_ceil(2)); - let mut ticker = tokio::time::interval(deadline); + let mut ticker = tokio::time::interval_at(tokio::time::Instant::now() + deadline, deadline); ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); const MAX_CONSECUTIVE_MISSES: u32 = 3;