diff --git a/docs/docs/users/reference/env_variables.md b/docs/docs/users/reference/env_variables.md index eab228b3e6f..fbb191370ba 100644 --- a/docs/docs/users/reference/env_variables.md +++ b/docs/docs/users/reference/env_variables.md @@ -77,6 +77,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/docs/docs/users/reference/metrics.md b/docs/docs/users/reference/metrics.md index b8f20c6d845..eb47fd6c4d2 100644 --- a/docs/docs/users/reference/metrics.md +++ b/docs/docs/users/reference/metrics.md @@ -22,6 +22,7 @@ title: Metrics | `bitswap_message_count` | Counter | Count | Number of `bitswap` messages. Indexed by `type` | | `bitswap_container_capacities` | Gauge | Count | Capacity for each `bitswap` container. Indexed by `type` | | `bitswap_get_block_time` | Histogram | Seconds | Duration of `get_block` | +| `drand_gossip_entries_total` | Counter | Count | drand entries received over gossipsub. Indexed by `verdict` and `reason` | | `mpool_message_total` | Gauge | Count | Total number of messages in the message pool | | `build_info` | Gauge | N/A | Semantic version of the forest binary. Indexed by `version` | | `process_start_time_seconds` | Gauge | Seconds | Time that the process started (in seconds since the UNIX epoch) | diff --git a/src/beacon/drand.rs b/src/beacon/drand.rs index 49d0407b5ee..3fb5e7c51b7 100644 --- a/src/beacon/drand.rs +++ b/src/beacon/drand.rs @@ -140,6 +140,14 @@ impl BeaconSchedule { } } + /// The first unchained beacon in schedule order. + pub fn unchained_beacon(&self) -> Option<&DrandBeacon> { + self.0.iter().find_map(|point| match &point.beacon { + BeaconImpl::Drand(beacon) if beacon.network().is_unchained() => Some(beacon), + _ => None, + }) + } + 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 @@ -200,6 +208,11 @@ pub trait Beacon { None } + /// Chain hash identifying the `drand` network, if any. + fn chain_hash(&self) -> Option<&str> { + None + } + /// Fetches `round`, waiting only until that `drand` round is produced (per /// [`Self::beacon_round_timestamp`]) rather than until the later Filecoin epoch. async fn entry_when_available(&self, round: u64) -> anyhow::Result { @@ -307,6 +320,11 @@ impl DrandBeacon { ); } } + + #[cfg(test)] + pub(crate) fn is_cached(&self, round: u64) -> bool { + self.verified_beacons.get(&round).is_some() + } } impl Beacon for DrandBeacon { @@ -483,4 +501,8 @@ impl Beacon for DrandBeacon { .saturating_add(round.saturating_sub(1).saturating_mul(self.interval)), ) } + + fn chain_hash(&self) -> Option<&str> { + Some(&self.hash) + } } diff --git a/src/beacon/mod.rs b/src/beacon/mod.rs index 30d85e2f5a1..72877fb4226 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -1,15 +1,15 @@ // Copyright 2019-2026 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT -pub mod beacon_entries; mod drand; -// Only test fixtures decode it until gossiped entries are validated. -#[cfg(test)] mod drand_pb; + +pub mod beacon_entries; pub mod metrics; 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/beacon/signatures/mod.rs b/src/beacon/signatures/mod.rs index c2089b86a4f..852e3e9d8d6 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(in crate::beacon) 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_"; /// Signs `message` on G1 with `secret`, as an unchained `drand` network signs a round. diff --git a/src/daemon/mod.rs b/src/daemon/mod.rs index 82064cfc0f1..905a75c86ab 100644 --- a/src/daemon/mod.rs +++ b/src/daemon/mod.rs @@ -19,7 +19,7 @@ use crate::cli_shared::{ use crate::daemon::{context::AppContext, db_util::import_chain_as_forest_car}; use crate::db::gc::SnapshotGarbageCollector; use crate::db::ttl::EthMappingCollector; -use crate::libp2p::{Libp2pService, PeerManager}; +use crate::libp2p::{DrandGossip, Libp2pService, PeerManager}; use crate::message_pool::{MessagePool, MpoolConfig, MpoolLocker, NonceTracker}; use crate::networks::{self, ChainConfig}; use crate::prelude::*; @@ -302,6 +302,8 @@ async fn create_p2p_service( let peer_manager = Arc::new(PeerManager::default()); services.spawn(peer_manager.clone().peer_operation_event_loop_task()); + // The state manager's schedule, so gossiped entries fill the cache block validation reads. + let drand_gossip = DrandGossip::new(ctx.state_manager.beacon_schedule().clone()); // Libp2p service setup let p2p_service = Libp2pService::new( config.network.clone(), @@ -310,6 +312,7 @@ async fn create_p2p_service( ctx.net_keypair.clone(), config.chain.genesis_name(), *ctx.state_manager.chain_store().genesis_block_header().cid(), + drand_gossip, ) .await?; Ok(p2p_service) diff --git a/src/libp2p/drand/metrics.rs b/src/libp2p/drand/metrics.rs new file mode 100644 index 00000000000..2b90c2d9148 --- /dev/null +++ b/src/libp2p/drand/metrics.rs @@ -0,0 +1,67 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use std::sync::LazyLock; + +use libp2p::gossipsub::MessageAcceptance; +use prometheus_client::{ + encoding::EncodeLabelSet, + metrics::{counter::Counter, family::Family}, +}; + +/// Why a gossiped `drand` entry got its verdict. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Reason { + Verified, + InvalidSignature, + Malformed, + Future, + Stale, + NoSlot, + NoBeacon, +} + +impl Reason { + fn as_str(self) -> &'static str { + match self { + Self::Verified => "verified", + Self::InvalidSignature => "invalid_signature", + Self::Malformed => "malformed", + Self::Future => "future", + Self::Stale => "stale", + Self::NoSlot => "no_slot", + Self::NoBeacon => "no_beacon", + } + } +} + +#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet)] +struct VerdictLabels { + verdict: &'static str, + reason: &'static str, +} + +// Exported as `drand_gossip_entries_total`. +static DRAND_GOSSIP_ENTRIES: LazyLock> = LazyLock::new(|| { + let metric = Family::default(); + crate::metrics::default_registry().register( + "drand_gossip_entries", + "drand beacon entries received over gossipsub, by validation verdict and reason", + metric.clone(), + ); + metric +}); + +pub(super) fn record(acceptance: &MessageAcceptance, reason: Reason) { + let verdict = match acceptance { + MessageAcceptance::Accept => "accept", + MessageAcceptance::Reject => "reject", + MessageAcceptance::Ignore => "ignore", + }; + DRAND_GOSSIP_ENTRIES + .get_or_create(&VerdictLabels { + verdict, + reason: reason.as_str(), + }) + .inc(); +} diff --git a/src/libp2p/drand/mod.rs b/src/libp2p/drand/mod.rs new file mode 100644 index 00000000000..384343ff992 --- /dev/null +++ b/src/libp2p/drand/mod.rs @@ -0,0 +1,54 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +mod metrics; +#[cfg(test)] +pub(in crate::libp2p) mod test_utils; +mod validator; + +pub(in crate::libp2p) use validator::DrandGossipValidator; + +use std::sync::Arc; + +use libp2p::{ + PeerId, + gossipsub::{MessageId, TopicHash}, +}; + +use super::service::Verdict; +use crate::beacon::{Beacon as _, BeaconSchedule}; + +/// `gossipsub` topic prefix of a `drand` network, followed by its chain hash. +const PUBSUB_DRAND_STR: &str = "/drand/pubsub/v0.0.0"; + +/// The unchained beacon's `drand` topic and its validator. +pub struct DrandGossip { + pub(in crate::libp2p) topic: TopicHash, + pub(in crate::libp2p) verdicts: flume::Receiver, + validator: DrandGossipValidator, +} + +impl DrandGossip { + /// `None` without an unchained beacon. `schedule` must be the one block validation uses, as + /// gossip fills its cache. + pub fn new(schedule: Arc) -> Option { + let beacon = schedule.unchained_beacon()?; + let topic = TopicHash::from_raw(format!("{PUBSUB_DRAND_STR}/{}", beacon.chain_hash()?)); + let (verdicts_tx, verdicts) = flume::unbounded(); + Some(Self { + topic, + verdicts, + validator: DrandGossipValidator::new(schedule, verdicts_tx), + }) + } + + /// Validates a topic message; the verdict arrives on [`Self::verdicts`]. + pub(in crate::libp2p) fn on_message(&self, message_id: MessageId, source: PeerId, data: &[u8]) { + self.validator + .handle(data, message_id, source, unix_now_secs()); + } +} + +fn unix_now_secs() -> u64 { + chrono::Utc::now().timestamp().max(0) as u64 +} diff --git a/src/libp2p/drand/test_utils.rs b/src/libp2p/drand/test_utils.rs new file mode 100644 index 00000000000..a51bfebf559 --- /dev/null +++ b/src/libp2p/drand/test_utils.rs @@ -0,0 +1,46 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use std::sync::Arc; + +use libp2p::gossipsub::MessageAcceptance; + +use crate::beacon::{ + Beacon as _, BeaconPoint, BeaconSchedule, tests::fake_drand_signer::FakeDrandSigner, +}; + +/// On the test chain, every epoch draws from a round ending in 2. +pub(in crate::libp2p) const EPOCH_ROUND: u64 = 12; + +pub(in crate::libp2p) fn fake_drand_schedule() -> (FakeDrandSigner, Arc) { + let drand = FakeDrandSigner::new(); + let beacon = drand.beacon(); + ( + drand, + Arc::new(BeaconSchedule(vec![BeaconPoint::new(0, beacon)])), + ) +} + +/// Unix time right after `round` is produced by the fake `drand` chain. +pub(in crate::libp2p) fn time_of(schedule: &BeaconSchedule, round: u64) -> u64 { + schedule + .unchained_beacon() + .and_then(|beacon| beacon.beacon_round_timestamp(round)) + .expect("fake drand beacon has round timestamps") + + 1 +} + +pub(in crate::libp2p) fn is_cached(schedule: &BeaconSchedule, round: u64) -> bool { + schedule + .unchained_beacon() + .is_some_and(|beacon| beacon.is_cached(round)) +} + +// `MessageAcceptance` doesn't implement `PartialEq`. +pub(in crate::libp2p) fn assert_verdict(actual: &MessageAcceptance, expected: MessageAcceptance) { + assert_eq!( + std::mem::discriminant(actual), + std::mem::discriminant(&expected), + "verdict {actual:?}, expected {expected:?}" + ); +} diff --git a/src/libp2p/drand/validator.rs b/src/libp2p/drand/validator.rs new file mode 100644 index 00000000000..d7068eec901 --- /dev/null +++ b/src/libp2p/drand/validator.rs @@ -0,0 +1,239 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use std::sync::{Arc, LazyLock}; + +use libp2p::{ + PeerId, + gossipsub::{MessageAcceptance, MessageId}, +}; +use nonzero_ext::nonzero; +use quick_protobuf::{BytesReader, MessageRead as _}; +use tokio::sync::Semaphore; +use tracing::debug; + +use super::{ + Verdict, + metrics::{self, Reason}, +}; +use crate::beacon::{Beacon as _, BeaconEntry, BeaconSchedule, PublicRandResponse}; +use crate::utils::misc::env::env_or_default_logged; + +/// Older gossiped rounds are ignored; accepting them would let peers fill the cache. +pub(in crate::libp2p) const MAX_ROUND_AGE_SECS: u64 = 30; + +/// Rounds further ahead are ignored, not rejected: a slow local clock must not penalize peers. +pub(in crate::libp2p) const MAX_CLOCK_SKEW_SECS: u64 = 3; + +/// Concurrency cap for `drand` entries verified from `gossipsub`. Excess is dropped, not queued. +static MAX_CONCURRENT_VERIFICATIONS: LazyLock = LazyLock::new(|| { + env_or_default_logged( + "FOREST_MAX_CONCURRENT_DRAND_VERIFICATIONS", + nonzero!(4_usize), + ) + .get() + .min(Semaphore::MAX_PERMITS) +}); + +/// Verifies gossiped `drand` entries, reporting a verdict for each. Only malformed payloads and bad +/// signatures are rejected; anything else unverified is ignored. +pub(in crate::libp2p) struct DrandGossipValidator { + schedule: Arc, + limiter: Arc, + verdicts: flume::Sender, +} + +impl DrandGossipValidator { + pub(in crate::libp2p) fn new( + schedule: Arc, + verdicts: flume::Sender, + ) -> Self { + Self::with_max_verifications(schedule, verdicts, *MAX_CONCURRENT_VERIFICATIONS) + } + + pub(in crate::libp2p) fn with_max_verifications( + schedule: Arc, + verdicts: flume::Sender, + max_verifications: usize, + ) -> Self { + Self { + schedule, + limiter: Arc::new(Semaphore::new(max_verifications)), + verdicts, + } + } + + /// Validates a `PublicRandResponse` payload received at Unix time `now`, sending exactly one + /// [`Verdict`]. + pub(in crate::libp2p) fn handle( + &self, + data: &[u8], + message_id: MessageId, + source: PeerId, + now: u64, + ) { + let verdict = |acceptance, reason| { + report( + &self.verdicts, + message_id.clone(), + source, + acceptance, + reason, + ) + }; + let mut reader = BytesReader::from_bytes(data); + let Ok(response) = PublicRandResponse::from_reader(&mut reader, data) else { + debug!(%source, "drand entry could not be decoded"); + return verdict(MessageAcceptance::Reject, Reason::Malformed); + }; + let entry = BeaconEntry::new(response.round, response.signature); + let round = entry.round(); + + let Some(beacon) = self.schedule.unchained_beacon() else { + return verdict(MessageAcceptance::Ignore, Reason::NoBeacon); + }; + if round == 0 || entry.signature().is_empty() { + return verdict(MessageAcceptance::Reject, Reason::Malformed); + } + if let Some(round_ts) = beacon.beacon_round_timestamp(round) { + if round_ts > now.saturating_add(MAX_CLOCK_SKEW_SECS) { + debug!(round, %source, "ignoring drand entry: round is in the future"); + return verdict(MessageAcceptance::Ignore, Reason::Future); + } + if round_ts.saturating_add(MAX_ROUND_AGE_SECS) < now { + debug!(round, %source, "ignoring drand entry: round is stale"); + return verdict(MessageAcceptance::Ignore, Reason::Stale); + } + } + + let Ok(permit) = Arc::clone(&self.limiter).try_acquire_owned() else { + debug!( + round, + "ignoring drand entry: too many verifications in flight" + ); + return verdict(MessageAcceptance::Ignore, Reason::NoSlot); + }; + let schedule = Arc::clone(&self.schedule); + let verdicts = self.verdicts.clone(); + tokio::task::spawn_blocking(move || { + let _permit = permit; + let verified = schedule.unchained_beacon().is_some_and(|beacon| { + matches!( + beacon.verify_entries(std::slice::from_ref(&entry), &BeaconEntry::default()), + Ok(true) + ) + }); + let (acceptance, reason) = if verified { + debug!(round, "verified drand entry from gossipsub"); + (MessageAcceptance::Accept, Reason::Verified) + } else { + debug!(round, %source, "received invalid drand entry over gossipsub"); + (MessageAcceptance::Reject, Reason::InvalidSignature) + }; + report(&verdicts, message_id, source, acceptance, reason); + }); + } +} + +fn report( + verdicts: &flume::Sender, + message_id: MessageId, + source: PeerId, + acceptance: MessageAcceptance, + reason: Reason, +) { + metrics::record(&acceptance, reason); + // Fails only once the service is gone. + let _ = verdicts.send(Verdict { + message_id, + source, + acceptance, + }); +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::*; + use crate::beacon::tests::fake_drand_signer::FakeDrandSigner; + use crate::libp2p::drand::test_utils::{ + EPOCH_ROUND, assert_verdict, fake_drand_schedule, is_cached, time_of, + }; + + /// Runs `payload` through a validator with `max_verifications` slots. + async fn verdict_for( + schedule: &Arc, + payload: &[u8], + now: u64, + max_verifications: usize, + ) -> MessageAcceptance { + let (verdicts_tx, verdicts) = flume::unbounded(); + let validator = DrandGossipValidator::with_max_verifications( + schedule.clone(), + verdicts_tx, + max_verifications, + ); + validator.handle(payload, MessageId::new(b"drand"), PeerId::random(), now); + tokio::time::timeout(Duration::from_secs(5), verdicts.recv_async()) + .await + .expect("no verdict reported in time") + .expect("verdict channel closed") + .acceptance + } + + /// Valid epoch rounds are accepted and cached. + #[rstest::rstest] + #[case::epoch_round(EPOCH_ROUND, 0)] + // Within the clock skew allowance. + #[case::produced_slightly_ahead(EPOCH_ROUND, -(MAX_CLOCK_SKEW_SECS as i64))] + #[tokio::test] + async fn valid_round_is_accepted(#[case] round: u64, #[case] now_offset: i64) { + let (drand, schedule) = fake_drand_schedule(); + let now = time_of(&schedule, round).saturating_add_signed(now_offset); + + let verdict = verdict_for(&schedule, &drand.encode(round), now, 1).await; + + assert_verdict(&verdict, MessageAcceptance::Accept); + assert!(is_cached(&schedule, round), "round {round}"); + } + + /// Entries rejected or ignored without being cached. + #[rstest::rstest] + #[case::undecodable(|_: &FakeDrandSigner| vec![0xff, 0xff, 0xff], 0, MessageAcceptance::Reject)] + #[case::round_zero(|d: &FakeDrandSigner| d.encode_raw(0, d.entry(1).signature()), 0, MessageAcceptance::Reject)] + #[case::empty_signature(|d: &FakeDrandSigner| d.encode_raw(EPOCH_ROUND, &[]), 0, MessageAcceptance::Reject)] + // An epoch round with the next round's signature. + #[case::forged_signature( + |d: &FakeDrandSigner| d.encode_raw(EPOCH_ROUND, d.entry(EPOCH_ROUND + 1).signature()), + 0, + MessageAcceptance::Reject + )] + // Beyond the clock skew allowance. + #[case::future_round(|d: &FakeDrandSigner| d.encode(EPOCH_ROUND + 2), 0, MessageAcceptance::Ignore)] + #[case::stale_round(|d: &FakeDrandSigner| d.encode(EPOCH_ROUND), MAX_ROUND_AGE_SECS as i64, MessageAcceptance::Ignore)] + #[tokio::test] + async fn invalid_or_untimely_entry_is_not_accepted( + #[case] payload: fn(&FakeDrandSigner) -> Vec, + #[case] now_offset: i64, + #[case] expected: MessageAcceptance, + ) { + let (drand, schedule) = fake_drand_schedule(); + let now = time_of(&schedule, EPOCH_ROUND).saturating_add_signed(now_offset); + + let verdict = verdict_for(&schedule, &payload(&drand), now, 1).await; + + assert_verdict(&verdict, expected); + assert!(!is_cached(&schedule, EPOCH_ROUND), "nothing cached"); + } + + #[tokio::test] + async fn entry_ignored_when_verifications_saturated() { + let (drand, schedule) = fake_drand_schedule(); + let now = time_of(&schedule, EPOCH_ROUND); + + let verdict = verdict_for(&schedule, &drand.encode(EPOCH_ROUND), now, 0).await; + + assert_verdict(&verdict, MessageAcceptance::Ignore); + } +} diff --git a/src/libp2p/mod.rs b/src/libp2p/mod.rs index d8a01878033..3897bba24dd 100644 --- a/src/libp2p/mod.rs +++ b/src/libp2p/mod.rs @@ -5,6 +5,7 @@ mod behaviour; pub mod chain_exchange; mod config; pub mod discovery; +mod drand; mod gossip_params; pub mod hello; pub mod keypair; @@ -21,10 +22,11 @@ pub use libp2p::{ }; pub(in crate::libp2p) use self::behaviour::*; -pub use self::{config::*, peer_manager::*, service::*}; +pub use self::{config::*, drand::DrandGossip, peer_manager::*, service::*}; #[cfg(test)] mod tests { mod decode_test; + mod drand_gossip_tests; mod gossip_validation_test; mod gossipsub_filter_test; } diff --git a/src/libp2p/service.rs b/src/libp2p/service.rs index b992a4cd19a..742e2f18792 100644 --- a/src/libp2p/service.rs +++ b/src/libp2p/service.rs @@ -38,6 +38,7 @@ use super::{ ForestBehaviour, ForestBehaviourEvent, Libp2pConfig, chain_exchange::{ChainExchangeRequest, ChainExchangeResponse, make_chain_exchange_response}, discovery::{DerivedDiscoveryBehaviourEvent, PeerInfo}, + drand::DrandGossip, }; use crate::libp2p::{ PeerManager, PeerOperation, @@ -186,6 +187,7 @@ pub enum NetRPCMethods { pub struct Libp2pService { swarm: Swarm, bootstrap_peers: HashMap, + drand: Option, cs: ChainStore, peer_manager: Arc, network_receiver_in: flume::Receiver, @@ -204,6 +206,7 @@ impl Libp2pService { net_keypair: Keypair, network_name: GenesisNetworkName, genesis_cid: Cid, + drand: Option, ) -> anyhow::Result { let behaviour = ForestBehaviour::new(&net_keypair, &config, &network_name, peer_manager.clone()) @@ -277,6 +280,7 @@ impl Libp2pService { Ok(Libp2pService { swarm, bootstrap_peers, + drand, cs, peer_manager, network_receiver_in, @@ -298,6 +302,12 @@ impl Libp2pService { warn!("Failed to bootstrap with Kademlia: {e:#}"); } + let mut drand_verdicts = match &self.drand { + Some(drand) => drand.verdicts.clone().into_stream().left_stream(), + None => futures::stream::pending().right_stream(), + } + .fuse(); + let bitswap_request_manager = self.swarm.behaviour().bitswap.request_manager(); let mut swarm_stream = self.swarm.fuse(); let mut network_stream = self.network_receiver_in.stream().fuse(); @@ -347,7 +357,8 @@ impl Libp2pService { &self.network_sender_out, cx_response_tx.clone(), &pubsub_block_str, - &pubsub_msg_str,).await; + &pubsub_msg_str, + self.drand.as_ref()).await; }, None => { break; }, _ => { }, @@ -397,6 +408,14 @@ impl Libp2pService { _ = bootstrap_peer_dialer_interval_stream.next() => { dial_to_bootstrap_peers_if_needed(swarm_stream.get_mut(), &self.bootstrap_peers); } + verdict = drand_verdicts.next() => { + if let Some(Verdict { message_id, source, acceptance }) = verdict { + swarm_stream + .get_mut() + .behaviour_mut() + .report_message_validation_result(&message_id, &source, acceptance); + } + } }; } Ok(()) @@ -660,12 +679,14 @@ pub(in crate::libp2p) struct Verdict { pub(in crate::libp2p) acceptance: MessageAcceptance, } -/// Decodes a `gossipsub` message, emits it and returns its verdict. +/// Decodes a `gossipsub` message and emits it. Returns its verdict, or `None` when the `drand` +/// module reports it later. pub(in crate::libp2p) async fn handle_gossip_event( e: gossipsub::Event, network_sender_out: &Sender, pubsub_block_str: &str, pubsub_msg_str: &str, + drand: Option<&DrandGossip>, ) -> Option { let gossipsub::Event::Message { propagation_source: source, @@ -675,10 +696,14 @@ pub(in crate::libp2p) async fn handle_gossip_event( else { return None; }; + trace!("Got a Gossip Message from {:?}", source); + if let Some(drand) = drand.filter(|drand| message.topic == drand.topic) { + drand.on_message(message_id, source, &message.data); + return None; + } + let topic = message.topic.as_str(); let message = message.data; - - trace!("Got a Gossip Message from {:?}", source); let acceptance = if topic == pubsub_block_str { match from_slice_with_fallback::(&message) { Ok(b) => { @@ -944,6 +969,7 @@ async fn handle_forest_behaviour_event( )>, pubsub_block_str: &str, pubsub_msg_str: &str, + drand: Option<&DrandGossip>, ) { match event { ForestBehaviourEvent::Discovery(discovery_out) => { @@ -960,8 +986,14 @@ async fn handle_forest_behaviour_event( message_id, source, acceptance, - }) = - handle_gossip_event(e, network_sender_out, pubsub_block_str, pubsub_msg_str).await + }) = handle_gossip_event( + e, + network_sender_out, + pubsub_block_str, + pubsub_msg_str, + drand, + ) + .await { swarm.behaviour_mut().report_message_validation_result( &message_id, diff --git a/src/libp2p/tests/drand_gossip_tests.rs b/src/libp2p/tests/drand_gossip_tests.rs new file mode 100644 index 00000000000..9e112311727 --- /dev/null +++ b/src/libp2p/tests/drand_gossip_tests.rs @@ -0,0 +1,37 @@ +// Copyright 2019-2026 ChainSafe Systems +// SPDX-License-Identifier: Apache-2.0, MIT + +use std::time::Duration; + +use super::gossip_validation_test::gossip_message_event; +use crate::libp2p::{ + drand::{ + DrandGossip, + test_utils::{EPOCH_ROUND, fake_drand_schedule}, + }, + service::handle_gossip_event, +}; + +/// `drand` verdicts are deferred to the `drand` module, and nothing reaches the node. +#[tokio::test] +async fn drand_gossip_is_handed_to_the_drand_module() { + let (drand, schedule) = fake_drand_schedule(); + let gossip = DrandGossip::new(schedule).unwrap(); + let (tx, rx) = flume::unbounded(); + + let verdict = handle_gossip_event( + gossip_message_event(drand.encode(EPOCH_ROUND), gossip.topic.clone()), + &tx, + "", + "", + Some(&gossip), + ) + .await; + + assert!(verdict.is_none(), "the drand verdict is deferred"); + assert!(rx.try_recv().is_err(), "nothing is emitted to the node"); + tokio::time::timeout(Duration::from_secs(5), gossip.verdicts.recv_async()) + .await + .expect("no verdict reported") + .unwrap(); +} diff --git a/src/libp2p/tests/gossip_validation_test.rs b/src/libp2p/tests/gossip_validation_test.rs index 588891108d0..677bd51a6cf 100644 --- a/src/libp2p/tests/gossip_validation_test.rs +++ b/src/libp2p/tests/gossip_validation_test.rs @@ -10,6 +10,7 @@ use crate::{ blocks::{CachingBlockHeader, GossipBlock, RawBlockHeader}, libp2p::{ NetworkEvent, PubsubMessage, PubsubTopic, + drand::test_utils::assert_verdict, service::{Verdict, handle_gossip_event}, }, message::SignedMessage, @@ -23,16 +24,7 @@ use crate::{ const NETWORK: &str = "testgossipvalidation"; -// `MessageAcceptance` doesn't implement `PartialEq`. -fn assert_verdict(actual: &MessageAcceptance, expected: MessageAcceptance) { - assert_eq!( - std::mem::discriminant(actual), - std::mem::discriminant(&expected), - "verdict {actual:?}, expected {expected:?}" - ); -} - -fn gossip_message_event(data: Vec, topic: TopicHash) -> gossipsub::Event { +pub(super) fn gossip_message_event(data: Vec, topic: TopicHash) -> gossipsub::Event { gossipsub::Event::Message { propagation_source: PeerId::random(), message_id: MessageId::new(b"test"), @@ -112,6 +104,7 @@ async fn filecoin_gossip_verdicts( &tx, &block_topic, &msg_topic, + None, ) .await .expect("an immediate verdict");