forest/libp2p/
behaviour.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4use std::{
5    num::NonZeroUsize,
6    sync::{Arc, LazyLock},
7};
8
9use super::{
10    PeerManager,
11    discovery::{DerivedDiscoveryBehaviourEvent, DiscoveryEvent, PeerInfo},
12};
13use crate::libp2p_bitswap::BitswapBehaviour;
14use crate::utils::{encoding::blake2b_256, version::FOREST_VERSION_STRING};
15use crate::{
16    libp2p::{
17        chain_exchange::ChainExchangeBehaviour,
18        config::Libp2pConfig,
19        discovery::{DiscoveryBehaviour, DiscoveryConfig},
20        gossip_params::{build_peer_score_params, build_peer_score_threshold},
21        hello::HelloBehaviour,
22    },
23    networks::GenesisNetworkName,
24};
25use ahash::{HashMap, HashSet};
26use libp2p::{
27    Multiaddr, allow_block_list, connection_limits,
28    gossipsub::{
29        self, IdentTopic as Topic, MaxCountSubscriptionFilter, MessageAuthenticity, MessageId,
30        PublishError, SubscriptionError, ValidationMode, WhitelistSubscriptionFilter,
31    },
32    identity::{Keypair, PeerId},
33    kad::QueryId,
34    metrics::{Metrics, Recorder},
35    ping, request_response,
36    swarm::NetworkBehaviour,
37};
38use tracing::info;
39
40/// Libp2p behavior for the Forest node. This handles all sub protocols needed
41/// for a Filecoin node.
42#[derive(NetworkBehaviour)]
43pub(in crate::libp2p) struct ForestBehaviour {
44    // Behaviours that manage connections should come first, to get rid of some panics in debug build.
45    // See <https://github.com/libp2p/rust-libp2p/issues/4773#issuecomment-2042676966>
46    connection_limits: connection_limits::Behaviour,
47    pub(super) blocked_peers: allow_block_list::Behaviour<allow_block_list::BlockedPeers>,
48    pub(super) discovery: DiscoveryBehaviour,
49    ping: ping::Behaviour,
50    gossipsub: Gossipsub,
51    pub(super) hello: HelloBehaviour,
52    pub(super) chain_exchange: ChainExchangeBehaviour,
53    pub(super) bitswap: BitswapBehaviour,
54}
55
56impl Recorder<ForestBehaviourEvent> for Metrics {
57    fn record(&self, event: &ForestBehaviourEvent) {
58        match event {
59            ForestBehaviourEvent::Gossipsub(e) => self.record(e),
60            ForestBehaviourEvent::Ping(ping_event) => self.record(ping_event),
61            ForestBehaviourEvent::Discovery(DiscoveryEvent::Discovery(e)) => match e.as_ref() {
62                DerivedDiscoveryBehaviourEvent::Identify(e) => self.record(e),
63                DerivedDiscoveryBehaviourEvent::Kademlia(e) => self.record(e),
64                _ => {}
65            },
66            _ => {}
67        }
68    }
69}
70
71pub(in crate::libp2p) type Gossipsub = gossipsub::Behaviour<
72    gossipsub::IdentityTransform,
73    MaxCountSubscriptionFilter<WhitelistSubscriptionFilter>,
74>;
75
76// Matches Lotus:
77// <https://github.com/filecoin-project/lotus/blob/558e55b0276ca8a593f84c997f4fc12eee24579b/node/modules/lp2p/pubsub.go#L386-L389>
78const MAX_SUBSCRIPTIONS_PER_REQUEST: usize = 100;
79
80/// Filter accepting only Forest's topics, bounded in count and per request.
81pub(in crate::libp2p) fn build_subscription_filter(
82    network_name: &GenesisNetworkName,
83) -> MaxCountSubscriptionFilter<WhitelistSubscriptionFilter> {
84    let allowed: Vec<_> = crate::libp2p::pubsub_topics(network_name)
85        .map(|t| t.hash())
86        .collect();
87    MaxCountSubscriptionFilter {
88        // Whitelisted topics are the only ones counted, so their number is an
89        // exact, self-maintaining bound.
90        max_subscribed_topics: allowed.len(),
91        max_subscriptions_per_request: MAX_SUBSCRIPTIONS_PER_REQUEST,
92        filter: WhitelistSubscriptionFilter(allowed.into_iter().collect()),
93    }
94}
95
96pub(in crate::libp2p) fn build_gossipsub(
97    local_key: &Keypair,
98    network_name: &GenesisNetworkName,
99) -> anyhow::Result<Gossipsub> {
100    let mut gs_config_builder = gossipsub::ConfigBuilder::default();
101    gs_config_builder.max_transmit_size(1 << 20);
102    gs_config_builder.validation_mode(ValidationMode::Strict);
103    gs_config_builder.message_id_fn(|msg: &gossipsub::Message| {
104        let s = blake2b_256(&msg.data);
105        MessageId::from(s)
106    });
107
108    let gossipsub_config = gs_config_builder.build()?;
109    let mut gossipsub = Gossipsub::new_with_subscription_filter(
110        MessageAuthenticity::Signed(local_key.clone()),
111        gossipsub_config,
112        build_subscription_filter(network_name),
113    )
114    .map_err(anyhow::Error::msg)?;
115
116    gossipsub
117        .with_peer_score(
118            build_peer_score_params(network_name),
119            build_peer_score_threshold(),
120        )
121        .map_err(anyhow::Error::msg)?;
122
123    Ok(gossipsub)
124}
125
126impl ForestBehaviour {
127    pub async fn new(
128        local_key: &Keypair,
129        config: &Libp2pConfig,
130        network_name: &GenesisNetworkName,
131        peer_manager: Arc<PeerManager>,
132    ) -> anyhow::Result<Self> {
133        const MAX_ESTABLISHED_PER_PEER: u32 = 4;
134        static MAX_CONCURRENT_REQUEST_RESPONSE_STREAMS_PER_PEER: LazyLock<usize> = LazyLock::new(
135            || {
136                std::env::var("FOREST_MAX_CONCURRENT_REQUEST_RESPONSE_STREAMS_PER_PEER")
137                .ok()
138                .map(|it|
139                    it.parse::<NonZeroUsize>()
140                        .expect("Failed to parse the `FOREST_MAX_CONCURRENT_REQUEST_RESPONSE_STREAMS_PER_PEER` environment variable value, a positive integer is expected.")
141                        .get())
142                .unwrap_or(10)
143            },
144        );
145
146        let max_concurrent_request_response_streams = (config.target_peer_count as usize)
147            .saturating_mul(*MAX_CONCURRENT_REQUEST_RESPONSE_STREAMS_PER_PEER);
148
149        let gossipsub = build_gossipsub(local_key, network_name)?;
150
151        let bitswap = BitswapBehaviour::new(
152            &[
153                "/chain/ipfs/bitswap/1.2.0",
154                "/chain/ipfs/bitswap/1.1.0",
155                "/chain/ipfs/bitswap/1.0.0",
156                "/chain/ipfs/bitswap",
157            ],
158            request_response::Config::default()
159                .with_max_concurrent_streams(max_concurrent_request_response_streams),
160        );
161        crate::libp2p_bitswap::register_metrics(&mut crate::metrics::collector_registry());
162
163        let discovery = DiscoveryConfig::new(local_key.public(), network_name)
164            .with_mdns(config.mdns)
165            .with_kademlia(config.kademlia)
166            .with_user_defined(config.bootstrap_peers.clone())
167            .await?
168            .target_peer_count(u64::from(config.target_peer_count))
169            .finish()?;
170
171        let connection_limits = connection_limits::Behaviour::new(
172            connection_limits::ConnectionLimits::default()
173                .with_max_pending_incoming(Some(
174                    config
175                        .target_peer_count
176                        .saturating_mul(MAX_ESTABLISHED_PER_PEER),
177                ))
178                .with_max_pending_outgoing(Some(
179                    config
180                        .target_peer_count
181                        .saturating_mul(MAX_ESTABLISHED_PER_PEER),
182                ))
183                .with_max_established_incoming(Some(
184                    config
185                        .target_peer_count
186                        .saturating_mul(MAX_ESTABLISHED_PER_PEER),
187                ))
188                .with_max_established_outgoing(Some(
189                    config
190                        .target_peer_count
191                        .saturating_mul(MAX_ESTABLISHED_PER_PEER),
192                ))
193                .with_max_established_per_peer(Some(MAX_ESTABLISHED_PER_PEER)),
194        );
195
196        info!("libp2p Forest version: {}", FOREST_VERSION_STRING.as_str());
197        Ok(ForestBehaviour {
198            gossipsub,
199            discovery,
200            ping: Default::default(),
201            connection_limits,
202            blocked_peers: Default::default(),
203            bitswap,
204            hello: HelloBehaviour::new(
205                request_response::Config::default()
206                    .with_max_concurrent_streams(max_concurrent_request_response_streams),
207                peer_manager,
208            ),
209            chain_exchange: ChainExchangeBehaviour::new(
210                request_response::Config::default()
211                    .with_max_concurrent_streams(max_concurrent_request_response_streams),
212            ),
213        })
214    }
215
216    /// Bootstrap Kademlia network
217    pub fn bootstrap(&mut self) -> anyhow::Result<QueryId> {
218        self.discovery.bootstrap()
219    }
220
221    /// Publish data over the gossip network.
222    pub fn publish(
223        &mut self,
224        topic: Topic,
225        data: impl Into<Vec<u8>>,
226    ) -> Result<MessageId, PublishError> {
227        self.gossipsub.publish(topic, data)
228    }
229
230    /// Subscribe to a gossip topic.
231    pub fn subscribe(&mut self, topic: &Topic) -> Result<bool, SubscriptionError> {
232        self.gossipsub.subscribe(topic)
233    }
234
235    /// Returns a set of peer ids
236    pub fn peers(&self) -> &HashSet<PeerId> {
237        self.discovery.peers()
238    }
239
240    /// Returns a map of peer ids and their multi-addresses
241    pub fn peer_addresses(&self) -> HashMap<PeerId, HashSet<Multiaddr>> {
242        self.discovery.peer_addresses()
243    }
244
245    pub fn peer_info(&self, peer_id: &PeerId) -> Option<&PeerInfo> {
246        self.discovery.peer_info(peer_id)
247    }
248}
forest/libp2p/behaviour.rs

forest/libp2p/
behaviour.rs
