feat(lip2p/drand): introduce drand libp2p module

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                                                                                                                      |

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)                              |

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<BeaconEntry> {
@@ -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)
+    }
 }

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;

src/beacon/signatures/mod.rs

@@ -13,7 +13,7 @@ use rayon::prelude::*;
 pub use bls_signatures::{PublicKey as PublicKeyOnG1, Signature as SignatureOnG2};
 
 // See <https://www.ietf.org/archive/id/draft-irtf-cfrg-bls-signature-05.html#name-basic>
-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.

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)

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<Family<VerdictLabels, Counter>> = 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();
+}

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<Verdict>,
+    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<BeaconSchedule>) -> Option<Self> {
+        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
+}

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<BeaconSchedule>) {
+    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:?}"
+    );
+}