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 new file mode 100644 index 000000000000..74319bd75a4b --- /dev/null +++ b/proto/drand_pb.proto @@ -0,0 +1,9 @@ +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/drand.rs b/src/beacon/drand.rs index bbf151b29366..1b1a3a9b7410 100644 --- a/src/beacon/drand.rs +++ b/src/beacon/drand.rs @@ -140,6 +140,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 diff --git a/src/beacon/drand_pb.rs b/src/beacon/drand_pb.rs new file mode 100644 index 000000000000..3447b19f9cb6 --- /dev/null +++ b/src/beacon/drand_pb.rs @@ -0,0 +1,55 @@ +// 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)] +#![allow(clippy::assigning_clones)] +#![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 e849804fb088..37fb5c73d722 100644 --- a/src/beacon/mod.rs +++ b/src/beacon/mod.rs @@ -3,10 +3,12 @@ pub mod beacon_entries; mod drand; +mod drand_pb; pub mod metrics; pub mod signatures; pub use beacon_entries::*; pub use drand::*; +pub use drand_pb::PublicRandResponse; #[cfg(test)] pub mod mock_beacon; @@ -15,4 +17,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..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 -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 new file mode 100644 index 000000000000..8240fc0f365d --- /dev/null +++ b/src/beacon/tests/fake_drand.rs @@ -0,0 +1,95 @@ +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 + // The fixture builds beacons directly; registering a collector here + // would clash with the one the real quicknet config registers. + register_metrics: false, + }, + } + } + + // 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 = + 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 e7d557a72bb0..f90a22f9a071 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::{ @@ -27,13 +28,13 @@ use crate::{ tipset_syncer::{TipsetSyncerError, validate_tipset}, validation::GossipBlockValidator, }, - libp2p::{NetworkEvent, PubsubMessage, hello::HelloRequest}, + libp2p::{NetworkEvent, NetworkMessage, PubsubMessage, PubsubTopic, hello::HelloRequest}, message_pool::MessagePool, networks::calculate_expected_epoch, prelude::*, shim::clock::ChainEpoch, state_manager::StateManager, - utils::misc::env::env_or_default_logged, + utils::{flume::FlumeSenderExt as _, misc::env::env_or_default_logged}, }; use arc_swap::ArcSwap; use chrono::Utc; @@ -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, @@ -228,6 +230,10 @@ 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(); let cancellation_token = CancellationToken::new(); let _cancellation_token_drop_guard = cancellation_token.drop_guard_ref(); @@ -243,6 +249,8 @@ 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(); + 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); @@ -309,6 +317,50 @@ async fn chain_follower( debug!("Received invalid GossipSub message: {}", why); } } + PubsubMessage::DrandEntry(entry) => { + 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; + }; + + if matches!( + beacon.verify_entries( + std::slice::from_ref(&entry), + &BeaconEntry::default() + ), + Ok(true) + ) { + info!( + round = entry.round(), + "verified drand entry from gossipsub" + ); + last_drand_entry.store( + Utc::now().timestamp().max(0) as u64, + Ordering::Relaxed, + ); + } else { + debug!( + round = entry.round(), + "received invalid drand entry over gossipsub" + ); + } + }); + } }, _ => {} } @@ -316,6 +368,17 @@ 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, network, last_drand_entry, cancellation_token) + .await; + } + }); + // Forward tipsets from miners into the state machine. set.spawn({ let state_changed = state_changed.clone(); @@ -473,6 +536,91 @@ async fn chain_follower( Ok(()) } +/// 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 fall back to fetching the beacon over HTTP. +async fn drand_gossip_watchdog( + state_manager: StateManager, + network: SyncNetworkContext, + last_drand_entry: Arc, + cancellation_token: CancellationToken, +) { + if state_manager.beacon_schedule().unchained_beacon().is_none() { + return; + } + + let deadline = + Duration::from_secs(u64::from(state_manager.chain_config().block_delay_secs).div_ceil(2)); + + 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; + + let mut stale = false; + let mut consecutive_misses = 0_u32; + + 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() { + consecutive_misses = 0; + 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; + } + }; + // 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; + 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)); + } + } +} + // Increment the gossipsub event metrics. fn inc_gossipsub_event_metrics(event: &NetworkEvent) { let label = match event { @@ -485,6 +633,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 @@ -594,6 +743,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( 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..8a2fd234444d 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_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, @@ -27,7 +27,7 @@ use libp2p::{ Multiaddr, allow_block_list, connection_limits, gossipsub::{ self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAuthenticity, MessageId, - PublishError, SubscriptionError, ValidationMode, WhitelistSubscriptionFilter, + PublishError, SubscriptionError, TopicHash, ValidationMode, WhitelistSubscriptionFilter, }, identity::{Keypair, PeerId}, kad::QueryId, @@ -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,15 +111,12 @@ 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_threshold(), - ) + .with_peer_score(build_peer_score_params(cfg), build_peer_score_threshold()) .map_err(anyhow::Error::msg)?; Ok(gossipsub) @@ -128,6 +127,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 +146,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", @@ -232,6 +230,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/gossip_params.rs b/src/libp2p/gossip_params.rs index 96a1ba54d1af..3da6dcc92d1b 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 @@ -80,18 +78,17 @@ fn build_block_topic_config() -> TopicScoreParams { } } -pub(in crate::libp2p) fn build_peer_score_params( - network_name: &GenesisNetworkName, -) -> 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(); - 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/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 f0ac4b257d07..2f5d16c0964a 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}; @@ -15,6 +16,7 @@ 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, @@ -30,11 +32,14 @@ use libp2p::{ tcp, yamux, }; use nonzero_ext::nonzero; +use quick_protobuf::{BytesReader, MessageRead as _}; + 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,31 +83,51 @@ 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 /// 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, #[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 { +pub fn pubsub_topics(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 +162,8 @@ pub enum PubsubMessage { Block(GossipBlock), /// Messages that come over the message topic Message(SignedMessage), + /// Messages that come over the `drand` topic + DrandEntry(BeaconEntry), } /// Messages into the service to handle. @@ -146,6 +173,7 @@ pub enum NetworkMessage { topic: IdentTopic, message: Vec, }, + ResubscribeTopic(PubsubTopic), ChainExchangeRequest { peer_id: PeerId, request: ChainExchangeRequest, @@ -191,8 +219,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 +232,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 +265,15 @@ 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}"))?; + info!("Subscribed to gossipsub topic {topic} ({kind:?})"); + 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; }, _ => { }, @@ -360,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; } }, @@ -463,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 } => { @@ -478,6 +522,27 @@ async fn handle_network_message( } } } + NetworkMessage::ResubscribeTopic(pubsub_topic) => { + for (topic_hash, kind) in pubsub_topic_kinds.iter() { + 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(); + + 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, @@ -654,8 +719,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 +727,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, @@ -680,24 +745,52 @@ async fn handle_gossip_event( Err(e) => { warn!("Gossip Block from peer {source:?} could not be deserialized: {e:#}",); } - } - } else if topic == pubsub_msg_str { - match from_slice_with_fallback::(&message) { - Ok(m) => { - emit_event( - network_sender_out, - NetworkEvent::PubsubMessage { - message: PubsubMessage::Message(m), - }, - ) - .await; + }, + 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:#}" + ); + } } - Err(e) => { - warn!("Gossip Message from peer {source:?} could not be deserialized: {e:#}"); + } + Some(PubsubTopic::Drand) => { + 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 { + message: PubsubMessage::DrandEntry(BeaconEntry::new( + r.round, + r.signature, + )), + }, + ) + .await; + } + Err(e) => { + warn!( + "Gossip drand entry from peer {source:?} could not be decoded: {e:#}" + ); + } } } - } else { - warn!("Getting gossip messages from unknown topic: {topic}"); + None => warn!("Getting gossip messages from unknown topic: {topic}"), } } } @@ -923,8 +1016,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 +1029,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/libp2p/tests/drand_gossip_tests.rs b/src/libp2p/tests/drand_gossip_tests.rs new file mode 100644 index 000000000000..58a3872a2814 --- /dev/null +++ b/src/libp2p/tests/drand_gossip_tests.rs @@ -0,0 +1,174 @@ +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::{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 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" + ); +} diff --git a/src/libp2p/tests/gossipsub_filter_test.rs b/src/libp2p/tests/gossipsub_filter_test.rs index 8b0d05ed0eec..774bbdde6480 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. +pub(in crate::libp2p) struct TopicCfgOwner { + network_name: GenesisNetworkName, + drand_chain_hashes: Vec, +} + +impl TopicCfgOwner { + pub(in crate::libp2p) fn new() -> Self { + Self { + network_name: NETWORK.into(), + drand_chain_hashes: vec![DRAND_HASH.to_string()], + } + } + + pub(in crate::libp2p) 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 { + let owner = TopicCfgOwner::new(); Swarm::new_ephemeral_tokio(|identity| { - build_gossipsub(&identity, &NETWORK.into()).expect("failed to build gossipsub") + 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 75ee05bb5d6d..95a1f5d8779a 100644 --- a/src/networks/mod.rs +++ b/src/networks/mod.rs @@ -1,6 +1,7 @@ // Copyright 2019-2026 ChainSafe Systems // SPDX-License-Identifier: Apache-2.0, MIT +use std::slice::Iter; use std::str::FromStr; use std::sync::LazyLock; @@ -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,