chore: bump rust-libp2p to 0.57

Cargo.toml

@@ -22,8 +22,8 @@ flume = "0.12"
 futures = "0.3"
 get-size2 = { version = "0.10", features = ["derive", "bytes", "hashbrown"] }
 hashlink = "0.11"
-libp2p = { version = "0.56", default-features = false }
-libp2p-swarm-test = { version = "0.6", default-features = false, features = ["tokio"] }
+libp2p = { version = "0.57", default-features = false }
+libp2p-swarm-test = { version = "0.7", default-features = false, features = ["tokio"] }
 multihash-codetable = { version = "0.2", features = ["blake2b", "blake2s", "blake3", "sha2", "sha3", "strobe"] }
 rust2go = "0.4"
 serde = { version = "1", default-features = false, features = ["derive", "rc"] }
@@ -114,7 +114,7 @@ git-version = "0.3"
 group = "0.13"
 hashbrown = "0.17"
 hashlink = { workspace = true }
-hickory-resolver = { version = "0.25", default-features = false, features = ["system-config", "tokio"] }
+hickory-resolver = { version = "0.26", default-features = false, features = ["system-config", "tokio"] }
 http = "1"
 human-repr = { version = "1", features = ["iec", "space"] }
 humantime = "2"
@@ -176,7 +176,9 @@ pathfinding = "4"
 pin-project-lite = "0.2"
 positioned-io = "0.3"
 pretty_assertions = "1"
-prometheus-client = "0.23"
+prometheus-client = "0.25"
+# Only for `kubert-prometheus-process`, which is still pinned to 0.23. Drop once it publishes a 0.25 build.
+prometheus-client-023 = { package = "prometheus-client", version = "0.23" }
 puruspe = "0.4"
 quick-protobuf = "0.8"
 quick-protobuf-codec = "0.3"

deny.toml

@@ -6,8 +6,6 @@ ignore = [
   "RUSTSEC-2022-0061", # parity-wasm is deprecated
   "RUSTSEC-2024-0436", # paste is unmaintained
   "RUSTSEC-2025-0141", # bincode is unmaintained
-  "RUSTSEC-2026-0118", # hickory-*, waiting for libp2p to update and release
-  "RUSTSEC-2026-0119", # hickory-*, waiting for libp2p to update and release
 ]
 
 [licenses]

src/chain_sync/metrics.rs

@@ -2,7 +2,7 @@
 // SPDX-License-Identifier: Apache-2.0, MIT
 
 use prometheus_client::{
-    encoding::{EncodeLabelKey, EncodeLabelSet, EncodeLabelValue, LabelSetEncoder},
+    encoding::EncodeLabelSet,
     metrics::{counter::Counter, family::Family, histogram::Histogram},
 };
 use std::sync::LazyLock;
@@ -55,25 +55,16 @@ pub static GOSSIP_BLOCK_REJECTED_TOTAL: LazyLock<Family<GossipRejectReasonLabel,
         metric
     });
 
-#[derive(Clone, Debug, Hash, PartialEq, Eq, derive_more::Constructor)]
-pub struct Libp2pMessageKindLabel(&'static str);
+#[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet, derive_more::Constructor)]
+pub struct Libp2pMessageKindLabel {
+    libp2p_message_kind: &'static str,
+}
 
 #[derive(Clone, Debug, Hash, PartialEq, Eq, EncodeLabelSet, derive_more::Constructor)]
 pub struct GossipRejectReasonLabel {
     pub reason: &'static str,
 }
 
-impl EncodeLabelSet for Libp2pMessageKindLabel {
-    fn encode(&self, mut encoder: LabelSetEncoder) -> Result<(), std::fmt::Error> {
-        let mut label_encoder = encoder.encode_label();
-        let mut label_key_encoder = label_encoder.encode_label_key()?;
-        EncodeLabelKey::encode(&"libp2p_message_kind", &mut label_key_encoder)?;
-        let mut label_value_encoder = label_key_encoder.encode_label_value()?;
-        EncodeLabelValue::encode(&self.0, &mut label_value_encoder)?;
-        label_value_encoder.finish()
-    }
-}
-
 pub mod values {
     use super::Libp2pMessageKindLabel;
 

src/daemon/mod.rs

@@ -816,7 +816,7 @@ pub(super) async fn start_services(
     rpc_stop_handle: jsonrpsee::server::StopHandle,
 ) -> anyhow::Result<()> {
     // Cleanup the collector prometheus metrics registry on start
-    crate::metrics::reset_collector_registry();
+    crate::metrics::reset_collector_registries();
     let mut services = JoinSet::new();
     let network = config.chain();
     let ctx = AppContext::init(opts, &config).await?;

src/libp2p/behaviour.rs

@@ -100,6 +100,8 @@ pub(in crate::libp2p) fn build_gossipsub(
     let mut gs_config_builder = gossipsub::ConfigBuilder::default();
     gs_config_builder.max_transmit_size(1 << 20);
     gs_config_builder.validation_mode(ValidationMode::Strict);
+    gs_config_builder.max_publish_messages(5000);
+    gs_config_builder.max_control_message_size(512 * 1024);
     gs_config_builder.message_id_fn(|msg: &gossipsub::Message| {
         let s = blake2b_256(&msg.data);
         MessageId::from(s)

src/libp2p/discovery.rs

@@ -10,6 +10,8 @@ use std::{
 };
 
 use ahash::{HashMap, HashMapExt, HashSet, HashSetExt};
+use anyhow::Context as _;
+use hickory_resolver::proto::rr::RData;
 use libp2p::{
     StreamProtocol, autonat,
     core::Multiaddr,
@@ -455,11 +457,17 @@ impl NetworkBehaviour for DiscoveryBehaviour {
                         }
                         DerivedDiscoveryBehaviourEvent::Autonat(_) => {}
                         DerivedDiscoveryBehaviourEvent::Upnp(ev) => match ev {
-                            upnp::Event::NewExternalAddr(addr) => {
-                                info!("UPnP NewExternalAddr: {addr}");
+                            upnp::Event::NewExternalAddr {
+                                local_addr,
+                                external_addr,
+                            } => {
+                                info!("UPnP NewExternalAddr: {local_addr} -> {external_addr}");
                             }
-                            upnp::Event::ExpiredExternalAddr(addr) => {
-                                info!("UPnP ExpiredExternalAddr: {addr}");
+                            upnp::Event::ExpiredExternalAddr {
+                                local_addr,
+                                external_addr,
+                            } => {
+                                info!("UPnP ExpiredExternalAddr: {local_addr} -> {external_addr}");
                             }
                             upnp::Event::GatewayNotFound => {
                                 info!("UPnP GatewayNotFound");
@@ -550,33 +558,43 @@ impl NetworkBehaviour for DiscoveryBehaviour {
 // Note: The function is async because the sync API `hickory_resolver::Resolver` is a wrapper of
 // the async API and does not work inside another tokio runtime
 async fn resolve_libp2p_dnsaddr(name: &str) -> anyhow::Result<Vec<(PeerId, Multiaddr)>> {
-    let resolver = hickory_resolver::TokioResolver::builder_tokio()?.build();
+    let resolver = hickory_resolver::TokioResolver::builder_tokio()
+        .context("failed to read the system DNS configuration")?
+        .build()
+        .context("failed to build the DNS resolver")?;
 
     let name = ["_dnsaddr.", name].concat();
-    let txts = resolver.txt_lookup(name).await?;
+    let lookup = resolver
+        .txt_lookup(name.as_str())
+        .await
+        .with_context(|| format!("TXT lookup failed for {name}"))?;
 
     let mut pairs = vec![];
-    for txt in txts {
-        if let Some(chars) = txt.txt_data().first() {
-            match parse_dnsaddr_txt(chars) {
-                Err(e) => {
-                    // Skip over seemingly invalid entries.
-                    tracing::debug!("Invalid TXT record: {:?}", e);
-                }
-                Ok(mut addr) => {
-                    if let Some(Protocol::P2p(peer_id)) = addr.pop() {
-                        pairs.push((peer_id, addr))
-                    } else {
-                        tracing::debug!("Failed to parse peer id from {addr}")
-                    }
+    for record in lookup.answers() {
+        let RData::TXT(txt) = &record.data else {
+            continue;
+        };
+        // A TXT RDATA may be published as several character-strings. go-libp2p joins them, so
+        // Forest must too or the same zone yields a different bootstrap set on each node.
+        let chars = txt.txt_data.concat();
+        match parse_dnsaddr_txt(&chars) {
+            Err(e) => {
+                // Skip over seemingly invalid entries.
+                tracing::warn!("Invalid TXT record: {e:?}");
+            }
+            Ok(mut addr) => {
+                if let Some(Protocol::P2p(peer_id)) = addr.pop() {
+                    pairs.push((peer_id, addr))
+                } else {
+                    tracing::warn!("Failed to parse peer id from {addr}")
                 }
             }
         }
     }
     Ok(pairs)
 }
 
-/// Parses a `<character-string>` of a `dnsaddr` `TXT` record.
+/// Parses the value of a `dnsaddr` `TXT` record.
 fn parse_dnsaddr_txt(txt: &[u8]) -> io::Result<Multiaddr> {
     let s = str::from_utf8(txt).map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
     match s.strip_prefix("dnsaddr=") {
@@ -601,6 +619,36 @@ mod tests {
     use libp2p_swarm_test::SwarmExt as _;
     use std::str::FromStr as _;
 
+    const DNSADDR_TXT: &str = "dnsaddr=/dns/bootstrap.filecoin.chain.love/tcp/1235/p2p/12D3KooWBF8cpp65hp2u9LK5mh19x67ftAam84z9LsfaquTDSBpt";
+
+    #[test]
+    fn parse_dnsaddr_txt_accepts_a_bootstrap_record() {
+        let addr = parse_dnsaddr_txt(DNSADDR_TXT.as_bytes()).unwrap();
+        assert!(matches!(addr.iter().last(), Some(Protocol::P2p(_))));
+    }
+
+    #[test]
+    fn parse_dnsaddr_txt_rejects_non_dnsaddr_records() {
+        parse_dnsaddr_txt(b"v=spf1 -all").unwrap_err();
+        parse_dnsaddr_txt(b"dnsaddr=not-a-multiaddr").unwrap_err();
+        parse_dnsaddr_txt(&[0xff]).unwrap_err();
+    }
+
+    /// A zone may publish one value as several character-strings; they must be joined without a
+    /// separator before parsing, matching go-libp2p.
+    #[test]
+    fn multi_segment_txt_record_is_joined() {
+        let (head, tail) = DNSADDR_TXT.split_at(60);
+        let split =
+            hickory_resolver::proto::rr::rdata::TXT::new(vec![head.to_string(), tail.to_string()]);
+
+        assert_eq!(split.txt_data.len(), 2);
+        assert_eq!(
+            parse_dnsaddr_txt(&split.txt_data.concat()).unwrap(),
+            parse_dnsaddr_txt(DNSADDR_TXT.as_bytes()).unwrap()
+        );
+    }
+
     #[tokio::test]
     async fn resolve_libp2p_dnsaddr_test() {
         async fn run() -> anyhow::Result<()> {

src/libp2p/rpc/mod.rs

@@ -4,7 +4,6 @@
 mod decoder;
 use std::{io, marker::PhantomData, time::Duration};
 
-use async_trait::async_trait;
 use decoder::DagCborDecodingReader;
 use futures::prelude::*;
 use libp2p::request_response::{self, OutboundFailure};
@@ -92,10 +91,9 @@ impl From<OutboundFailure> for RequestResponseError {
     }
 }
 
-#[async_trait]
 impl<P, RQ, RS, C> request_response::Codec for CborRequestResponse<P, RQ, RS, C>
 where
-    P: AsRef<str> + Send + Clone,
+    P: AsRef<str> + Send + Sync + Clone,
     RQ: Serialize + DeserializeOwned + Send + Sync,
     RS: Serialize + DeserializeOwned + Send + Sync,
     C: CodecConfig + Send + Sync,

src/libp2p_bitswap/internals/codec.rs

@@ -3,7 +3,6 @@
 
 use std::io;
 
-use async_trait::async_trait;
 use asynchronous_codec::{FramedRead, FramedWrite};
 use futures::{
     SinkExt, StreamExt,
@@ -32,7 +31,6 @@ fn codec() -> quick_protobuf_codec::Codec<bitswap_pb::Message> {
 #[derive(Default, Debug, Clone)]
 pub struct BitswapRequestResponseCodec;
 
-#[async_trait]
 impl request_response::Codec for BitswapRequestResponseCodec {
     type Protocol = &'static str;
     type Request = BitswapMessages;

src/metrics/mod.rs

@@ -28,6 +28,11 @@ static DEFAULT_REGISTRY: LazyLock<RwLock<prometheus_client::registry::Registry>>
 static COLLECTOR_REGISTRY: LazyLock<RwLock<prometheus_client::registry::Registry>> =
     LazyLock::new(Default::default);
 
+// Separate because `kubert-prometheus-process` is still on `prometheus-client` 0.23 while the rest
+// of the tree is on 0.25. Both encode to the same text format, so the scrape output is unaffected.
+static PROCESS_REGISTRY: LazyLock<RwLock<prometheus_client_023::registry::Registry>> =
+    LazyLock::new(Default::default);
+
 pub fn default_registry<'a>() -> RwLockWriteGuard<'a, prometheus_client::registry::Registry> {
     DEFAULT_REGISTRY.write()
 }
@@ -41,8 +46,21 @@ pub fn register_collector(collector: Box<dyn Collector>) {
     collector_registry().register_collector(collector)
 }
 
-pub fn reset_collector_registry() {
+fn register_process_metrics() {
+    if let Err(err) = kubert_prometheus_process::register(
+        PROCESS_REGISTRY.write().sub_registry_with_prefix("process"),
+    ) {
+        warn!("Failed to register process metrics: {err}");
+    }
+}
+
+/// Clears the collector and process registries, which are repopulated on each daemon run.
+///
+/// [`DEFAULT_REGISTRY`] is deliberately left alone: its families are registered from `LazyLock`
+/// statics, which only ever run once.
+pub fn reset_collector_registries() {
     *collector_registry() = Default::default();
+    *PROCESS_REGISTRY.write() = Default::default();
 }
 
 pub static RPC_METHOD_FAILURE: LazyLock<Family<RpcMethodLabel, Counter>> = LazyLock::new(|| {
@@ -89,12 +107,7 @@ pub async fn init_prometheus<DB>(
 where
     DB: DBStatistics + Send + Sync + 'static,
 {
-    // Add the process collector to the registry
-    if let Err(err) = kubert_prometheus_process::register(
-        collector_registry().sub_registry_with_prefix("process"),
-    ) {
-        warn!("Failed to register process metrics: {err}");
-    }
+    register_process_metrics();
 
     register_collector(Box::new(
         crate::utils::version::ForestVersionCollector::new(),
@@ -119,7 +132,7 @@ where
     Ok(axum::serve(prometheus_listener, app.into_make_service()).await?)
 }
 
-async fn collect_prometheus_metrics() -> impl IntoResponse {
+fn encode_metrics() -> String {
     let mut metrics = String::new();
     if let Err(e) =
         prometheus_client::encoding::text::encode_registry(&mut metrics, &DEFAULT_REGISTRY.read())
@@ -131,9 +144,20 @@ async fn collect_prometheus_metrics() -> impl IntoResponse {
     {
         warn!("failed to encode the collector metrics registry: {e}");
     };
+    if let Err(e) = prometheus_client_023::encoding::text::encode_registry(
+        &mut metrics,
+        &PROCESS_REGISTRY.read(),
+    ) {
+        warn!("failed to encode the process metrics registry: {e}");
+    };
     if let Err(e) = prometheus_client::encoding::text::encode_eof(&mut metrics) {
         warn!("failed to encode metrics eof {e}");
     };
+    metrics
+}
+
+async fn collect_prometheus_metrics() -> impl IntoResponse {
+    let metrics = encode_metrics();
     (
         StatusCode::OK,
         [("content-type", "text/plain; charset=utf-8")],
@@ -227,3 +251,40 @@ impl GaugeGuardExt for Gauge {
         GaugeGuard { gauge: self }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use itertools::Itertools as _;
+
+    /// `kubert-prometheus-process` registers into a registry from a different `prometheus-client`
+    /// version than the rest of the tree, so guard that its metrics still reach the scrape output.
+    #[test]
+    #[serial_test::serial]
+    fn process_metrics_are_scraped() {
+        *PROCESS_REGISTRY.write() = Default::default();
+        register_process_metrics();
+
+        let scraped = encode_metrics();
+
+        assert!(scraped.contains("process_start_time_seconds"));
+        assert!(scraped.contains("process_uptime_seconds_total"));
+        assert_eq!(scraped.matches("# EOF").count(), 1);
+        assert!(scraped.ends_with("# EOF\n"));
+
+        // A duplicate family makes the whole scrape unparseable, which the substring
+        // assertions above cannot see.
+        for prefix in ["# HELP ", "# TYPE "] {
+            let duplicates = scraped
+                .lines()
+                .filter_map(|l| l.strip_prefix(prefix))
+                .filter_map(|declaration| declaration.split_whitespace().next())
+                .duplicates()
+                .collect_vec();
+            assert!(
+                duplicates.is_empty(),
+                "duplicate `{prefix}` names: {duplicates:?}"
+            );
+        }
+    }
+}