forest/utils/
mod.rs

1// Copyright 2019-2026 ChainSafe Systems
2// SPDX-License-Identifier: Apache-2.0, MIT
3
4pub mod broadcast;
5pub mod cache;
6pub mod cid;
7pub mod db;
8pub mod encoding;
9pub mod flume;
10pub mod get_size;
11pub mod hash;
12pub mod io;
13pub mod misc;
14pub mod multihash;
15pub mod net;
16pub mod p2p;
17pub mod proofs_api;
18pub mod publisher;
19pub mod rand;
20pub mod reqwest_resume;
21mod shallow_clone;
22pub use shallow_clone::ShallowClone;
23#[cfg(feature = "sqlite")]
24pub mod sqlite;
25pub mod stats;
26pub mod stream;
27pub mod version;
28
29use anyhow::{Context as _, bail};
30use futures::Future;
31use libp2p::multiaddr::{Multiaddr, Protocol};
32use std::{str::FromStr, time::Duration};
33use tokio::time::sleep;
34use tracing::error;
35use url::Url;
36
37/// `"hunter2:/ip4/127.0.0.1/wss" -> "wss://:hunter2@127.0.0.1/"`
38#[derive(Clone, Debug)]
39pub struct UrlFromMultiAddr(pub Url);
40
41impl FromStr for UrlFromMultiAddr {
42    type Err = anyhow::Error;
43
44    fn from_str(s: &str) -> Result<Self, Self::Err> {
45        let (p, s) = match s.split_once(':') {
46            Some((first, rest)) => (Some(first), rest),
47            None => (None, s),
48        };
49        let m = Multiaddr::from_str(s).context("invalid multiaddr")?;
50        let mut u = multiaddr2url(&m).context("unsupported multiaddr")?;
51        if u.set_password(p).is_err() {
52            bail!("unsupported password")
53        }
54        Ok(Self(u))
55    }
56}
57
58/// `"/dns/example.com/tcp/8080/http" -> "http://example.com:8080/"`
59///
60/// Returns [`None`] on unsupported formats, or if there is a URL parsing error.
61///
62/// Note that [`Multiaddr`]s do NOT support a (URL) `path`, so that must be handled
63/// out-of-band.
64fn multiaddr2url(m: &Multiaddr) -> Option<Url> {
65    let mut components = m.iter().peekable();
66    let host = match components.next()? {
67        Protocol::Dns(it) | Protocol::Dns4(it) | Protocol::Dns6(it) | Protocol::Dnsaddr(it) => {
68            it.to_string()
69        }
70        Protocol::Ip4(it) => it.to_string(),
71        Protocol::Ip6(it) => it.to_string(),
72        _ => return None,
73    };
74    let port = match components.peek() {
75        Some(&Protocol::Tcp(port)) => {
76            components.next();
77            Some(port)
78        }
79        _ => None,
80    };
81    // ENHANCEMENT: could recognise `Tcp/443/Tls` as `https`
82    let scheme = match components.next()? {
83        Protocol::Http => "http",
84        Protocol::Https => "https",
85        Protocol::Ws(it) if it == "/" => "ws",
86        Protocol::Wss(it) if it == "/" => "wss",
87        _ => return None,
88    };
89    let None = components.next() else { return None };
90    let parse_me = match port {
91        Some(port) => format!("{scheme}://{host}:{port}"),
92        None => format!("{scheme}://{host}"),
93    };
94    parse_me.parse().ok()
95}
96
97#[test]
98fn test_url_from_multiaddr() {
99    #[track_caller]
100    fn do_test(input: &str, expected: &str) {
101        let UrlFromMultiAddr(url) = input.parse().unwrap();
102        assert_eq!(url.as_str(), expected, "input: {input}");
103    }
104    do_test("/dns/example.com/http", "http://example.com/");
105    do_test("/dns/example.com/tcp/8080/http", "http://example.com:8080/");
106    do_test("/dns/example.com/tcp/8081/ws", "ws://example.com:8081/");
107    do_test("/ip4/127.0.0.1/wss", "wss://127.0.0.1/");
108
109    // with password
110    do_test(
111        "hunter2:/dns/example.com/http",
112        "http://:hunter2@example.com/",
113    );
114    do_test(
115        "hunter2:/dns/example.com/tcp/8080/http",
116        "http://:hunter2@example.com:8080/",
117    );
118    do_test("hunter2:/ip4/127.0.0.1/wss", "wss://:hunter2@127.0.0.1/");
119}
120
121/// Keep running the future created by `make_fut` until the timeout or retry
122/// limit in `args` is reached.
123/// `F` _must_ be cancel safe.
124#[tracing::instrument(skip_all)]
125pub async fn retry<F, T, E>(
126    args: RetryArgs,
127    mut make_fut: impl FnMut() -> F,
128) -> Result<T, RetryError>
129where
130    F: Future<Output = Result<T, E>>,
131    E: std::fmt::Debug,
132{
133    let max_retries = args.max_retries.unwrap_or(usize::MAX);
134    let task = async {
135        for _ in 0..max_retries {
136            match make_fut().await {
137                Ok(ok) => return Ok(ok),
138                Err(err) => error!("retrying operation after {err:?}"),
139            }
140            if let Some(delay) = args.delay {
141                sleep(delay).await;
142            }
143        }
144        Err(RetryError::RetriesExceeded)
145    };
146
147    if let Some(timeout) = args.timeout {
148        tokio::time::timeout(timeout, task)
149            .await
150            .map_err(|_| RetryError::TimeoutExceeded)?
151    } else {
152        task.await
153    }
154}
155
156/// On timeout the blocking task is detached, not cancelled — a blocking syscall cannot
157/// be aborted, so anything the closure captures (e.g. a [`tempfile::TempPath`] whose
158/// drop removes the temporary file) is released only when the syscall eventually returns.
159pub async fn spawn_blocking_with_timeout<T: Send + 'static>(
160    timeout: Duration,
161    f: impl FnOnce() -> anyhow::Result<T> + Send + 'static,
162) -> anyhow::Result<T> {
163    tokio::time::timeout(timeout, tokio::task::spawn_blocking(f))
164        .await
165        .context("blocking operation timed out")??
166}
167
168#[derive(Debug, Clone, Copy, smart_default::SmartDefault)]
169pub struct RetryArgs {
170    #[default(Some(Duration::from_secs(1)))]
171    pub timeout: Option<Duration>,
172    #[default(Some(5))]
173    pub max_retries: Option<usize>,
174    #[default(Some(Duration::from_millis(200)))]
175    pub delay: Option<Duration>,
176}
177
178#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
179pub enum RetryError {
180    #[error("operation timed out")]
181    TimeoutExceeded,
182    #[error("retry limit exceeded")]
183    RetriesExceeded,
184}
185
186#[allow(dead_code)]
187#[cfg(test)]
188pub fn is_debug_build() -> bool {
189    cfg!(debug_assertions)
190}
191
192#[allow(dead_code)]
193#[cfg(test)]
194pub fn is_ci() -> bool {
195    // https://docs.github.com/en/actions/writing-workflows/choosing-what-your-workflow-does/store-information-in-variables#default-environment-variables
196    misc::env::is_env_truthy("CI")
197}
198
199#[cfg(test)]
200mod tests {
201    mod files;
202
203    use RetryError::{RetriesExceeded, TimeoutExceeded};
204    use futures::future::pending;
205    use std::{future::ready, sync::atomic::AtomicUsize};
206
207    use super::*;
208
209    impl RetryArgs {
210        fn new_ms(
211            timeout: impl Into<Option<u64>>,
212            max_retries: impl Into<Option<usize>>,
213            delay: impl Into<Option<u64>>,
214        ) -> Self {
215            Self {
216                timeout: timeout.into().map(Duration::from_millis),
217                max_retries: max_retries.into(),
218                delay: delay.into().map(Duration::from_millis),
219            }
220        }
221    }
222
223    #[tokio::test]
224    async fn timeout() {
225        let res = retry(RetryArgs::new_ms(1, None, None), pending::<Result<(), ()>>).await;
226        assert_eq!(Err(TimeoutExceeded), res);
227    }
228
229    #[tokio::test]
230    async fn retries() {
231        let res = retry(RetryArgs::new_ms(None, 1, None), || ready(Err::<(), _>(()))).await;
232        assert_eq!(Err(RetriesExceeded), res);
233    }
234
235    #[tokio::test]
236    async fn ok() {
237        let res = retry(RetryArgs::default(), || ready(Ok::<_, ()>(()))).await;
238        assert_eq!(Ok(()), res);
239    }
240
241    #[tokio::test]
242    async fn needs_retry() {
243        use std::sync::atomic::Ordering::SeqCst;
244        let count = AtomicUsize::new(0);
245        let res = retry(RetryArgs::new_ms(None, None, None), || async {
246            match count.fetch_add(1, SeqCst) > 5 {
247                true => Ok(()),
248                false => Err(()),
249            }
250        })
251        .await;
252        assert_eq!(Ok(()), res);
253        assert!(count.load(SeqCst) > 5);
254    }
255}
forest/utils/mod.rs

forest/utils/
mod.rs
