fedimint_core/util/
mod.rs

1pub mod backoff_util;
2/// Copied from `tokio_stream` 0.1.12 to use our optional Send bounds
3pub mod broadcaststream;
4#[cfg(feature = "uniffi")]
5pub mod ffi;
6pub mod update_merge;
7
8use std::convert::Infallible;
9use std::fmt::{Debug, Display, Formatter};
10use std::future::Future;
11use std::hash::Hash;
12use std::io::Write;
13use std::path::Path;
14use std::pin::Pin;
15use std::str::FromStr;
16use std::sync::LazyLock;
17use std::{fs, io};
18
19use anyhow::format_err;
20use fedimint_logging::LOG_CORE;
21pub use fedimint_util_error::*;
22use futures::StreamExt;
23use serde::{Deserialize, Serialize};
24use tokio::io::AsyncWriteExt;
25use tracing::{Instrument, Span, debug, warn};
26use url::{Host, ParseError, Url};
27
28use crate::envs::{FM_DEBUG_SHOW_SECRETS_ENV, is_env_var_set};
29use crate::task::MaybeSend;
30use crate::{apply, async_trait_maybe_send, maybe_add_send, runtime};
31
32/// Future that is `Send` unless targeting WASM
33pub type BoxFuture<'a, T> = Pin<Box<maybe_add_send!(dyn Future<Output = T> + 'a)>>;
34
35/// Stream that is `Send` unless targeting WASM
36pub type BoxStream<'a, T> = Pin<Box<maybe_add_send!(dyn futures::Stream<Item = T> + 'a)>>;
37
38#[apply(async_trait_maybe_send!)]
39pub trait NextOrPending {
40    type Output;
41
42    async fn next_or_pending(&mut self) -> Self::Output;
43
44    async fn ok(&mut self) -> anyhow::Result<Self::Output>;
45}
46
47#[apply(async_trait_maybe_send!)]
48impl<S> NextOrPending for S
49where
50    S: futures::Stream + Unpin + MaybeSend,
51    S::Item: MaybeSend,
52{
53    type Output = S::Item;
54
55    /// Waits for the next item in a stream. If the stream is closed while
56    /// waiting, returns an error.  Useful when expecting a stream to progress.
57    async fn ok(&mut self) -> anyhow::Result<Self::Output> {
58        self.next()
59            .await
60            .map_or_else(|| Err(format_err!("Stream was unexpectedly closed")), Ok)
61    }
62
63    /// Waits for the next item in a stream. If the stream is closed while
64    /// waiting the future will be pending forever. This is useful in cases
65    /// where the future will be cancelled by shutdown logic anyway and handling
66    /// each place where a stream may terminate would be too much trouble.
67    async fn next_or_pending(&mut self) -> Self::Output {
68        if let Some(item) = self.next().await {
69            item
70        } else {
71            debug!(target: LOG_CORE, "Stream ended in next_or_pending, pending forever to avoid throwing an error on shutdown");
72            std::future::pending().await
73        }
74    }
75}
76
77// TODO: make fully RFC1738 conformant
78/// Wrapper for `Url` that only prints the scheme, domain, port and path portion
79/// of a `Url` in its `Display` implementation.
80///
81/// This is useful to hide private
82/// information like user names and passwords in logs or UIs.
83///
84/// The output is not fully RFC1738 conformant but good enough for our current
85/// purposes.
86#[derive(Hash, Clone, Serialize, Deserialize, Eq, PartialEq, Ord, PartialOrd)]
87// nosemgrep: ban-raw-url
88pub struct SafeUrl(Url);
89
90#[cfg(feature = "uniffi")]
91uniffi::custom_type!(SafeUrl, String, {
92    lower: |u| u.0.to_string(),
93    try_lift: |s| SafeUrl::parse(&s).map_err(|e| anyhow::anyhow!("Invalid URL: {e}")),
94});
95
96impl SafeUrl {
97    pub fn parse(url_str: &str) -> Result<Self, ParseError> {
98        Url::parse(url_str).map(SafeUrl)
99    }
100
101    /// Warning: This removes the safety.
102    // nosemgrep: ban-raw-url
103    pub fn to_unsafe(self) -> Url {
104        self.0
105    }
106
107    #[allow(clippy::result_unit_err)] // just copying `url`'s API here
108    pub fn set_username(&mut self, username: &str) -> Result<(), ()> {
109        self.0.set_username(username)
110    }
111
112    #[allow(clippy::result_unit_err)] // just copying `url`'s API here
113    pub fn set_password(&mut self, password: Option<&str>) -> Result<(), ()> {
114        self.0.set_password(password)
115    }
116
117    #[allow(clippy::result_unit_err)] // just copying `url`'s API here
118    pub fn without_auth(&self) -> Result<Self, ()> {
119        let mut url = self.clone();
120
121        url.set_username("").and_then(|()| url.set_password(None))?;
122
123        Ok(url)
124    }
125
126    pub fn host(&self) -> Option<Host<&str>> {
127        self.0.host()
128    }
129    pub fn host_str(&self) -> Option<&str> {
130        self.0.host_str()
131    }
132    pub fn scheme(&self) -> &str {
133        self.0.scheme()
134    }
135    pub fn port(&self) -> Option<u16> {
136        self.0.port()
137    }
138    pub fn port_or_known_default(&self) -> Option<u16> {
139        self.0.port_or_known_default()
140    }
141    pub fn path(&self) -> &str {
142        self.0.path()
143    }
144    /// Warning: This will expose username & password if present.
145    pub fn as_str(&self) -> &str {
146        self.0.as_str()
147    }
148    pub fn username(&self) -> &str {
149        self.0.username()
150    }
151    pub fn password(&self) -> Option<&str> {
152        self.0.password()
153    }
154    pub fn join(&self, input: &str) -> Result<Self, ParseError> {
155        self.0.join(input).map(SafeUrl)
156    }
157
158    /// Append a relative path, ensuring exactly one `/` between
159    /// the base and the path segment.
160    ///
161    /// Unlike `Url::join` (RFC 3986), this never drops path
162    /// segments from the base — it always appends.
163    pub fn join_path(&self, path: &str) -> Self {
164        let base = self.to_string();
165        let base = base.trim_end_matches('/');
166        let path = path.trim_start_matches('/');
167        Self::parse(&format!("{base}/{path}"))
168            .expect("appending a relative path to a valid URL should produce a valid URL")
169    }
170
171    // It can be removed to use `is_onion_address()` implementation,
172    // once https://gitlab.torproject.org/tpo/core/arti/-/merge_requests/2214 lands.
173    #[allow(clippy::case_sensitive_file_extension_comparisons)]
174    pub fn is_onion_address(&self) -> bool {
175        let host = self.host_str().unwrap_or_default();
176
177        host.ends_with(".onion")
178    }
179
180    pub fn fragment(&self) -> Option<&str> {
181        self.0.fragment()
182    }
183
184    pub fn set_fragment(&mut self, arg: Option<&str>) {
185        self.0.set_fragment(arg);
186    }
187}
188
189static SHOW_SECRETS: LazyLock<bool> = LazyLock::new(|| {
190    let enable = is_env_var_set(FM_DEBUG_SHOW_SECRETS_ENV);
191
192    if enable {
193        warn!(target: LOG_CORE, "{} enabled. Please don't use in production.", FM_DEBUG_SHOW_SECRETS_ENV);
194    }
195
196    enable
197});
198
199impl Display for SafeUrl {
200    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
201        write!(f, "{}://", self.0.scheme())?;
202
203        if !self.0.username().is_empty() {
204            let show_secrets = *SHOW_SECRETS;
205            if show_secrets {
206                write!(f, "{}", self.0.username())?;
207            } else {
208                write!(f, "REDACTEDUSER")?;
209            }
210
211            if self.0.password().is_some() {
212                if show_secrets {
213                    write!(
214                        f,
215                        ":{}",
216                        self.0.password().expect("Just checked it's checked")
217                    )?;
218                } else {
219                    write!(f, ":REDACTEDPASS")?;
220                }
221            }
222
223            write!(f, "@")?;
224        }
225
226        if let Some(host) = self.0.host_str() {
227            write!(f, "{host}")?;
228        }
229
230        if let Some(port) = self.0.port() {
231            write!(f, ":{port}")?;
232        }
233
234        write!(f, "{}", self.0.path())?;
235
236        Ok(())
237    }
238}
239
240impl Debug for SafeUrl {
241    fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
242        write!(f, "SafeUrl(")?;
243        Display::fmt(self, f)?;
244        write!(f, ")")?;
245        Ok(())
246    }
247}
248
249impl From<Url> for SafeUrl {
250    fn from(u: Url) -> Self {
251        Self(u)
252    }
253}
254
255impl FromStr for SafeUrl {
256    type Err = ParseError;
257
258    #[inline]
259    fn from_str(input: &str) -> Result<Self, ParseError> {
260        Self::parse(input)
261    }
262}
263
264/// Write out a new file (like [`std::fs::write`] but fails if file already
265/// exists)
266#[cfg(not(target_family = "wasm"))]
267pub fn write_new<P: AsRef<Path>, C: AsRef<[u8]>>(path: P, contents: C) -> io::Result<()> {
268    let mut file = fs::File::options()
269        .write(true)
270        .create_new(true)
271        .open(path)?;
272    file.write_all(contents.as_ref())?;
273    file.sync_all()?;
274    Ok(())
275}
276
277#[cfg(not(target_family = "wasm"))]
278pub fn write_overwrite<P: AsRef<Path>, C: AsRef<[u8]>>(path: P, contents: C) -> io::Result<()> {
279    fs::File::options()
280        .write(true)
281        .create(true)
282        .truncate(true)
283        .open(path)?
284        .write_all(contents.as_ref())
285}
286
287#[cfg(not(target_family = "wasm"))]
288pub async fn write_overwrite_async<P: AsRef<Path>, C: AsRef<[u8]>>(
289    path: P,
290    contents: C,
291) -> io::Result<()> {
292    tokio::fs::OpenOptions::new()
293        .write(true)
294        .create(true)
295        .truncate(true)
296        .open(path)
297        .await?
298        .write_all(contents.as_ref())
299        .await
300}
301
302#[cfg(not(target_family = "wasm"))]
303pub async fn write_new_async<P: AsRef<Path>, C: AsRef<[u8]>>(
304    path: P,
305    contents: C,
306) -> io::Result<()> {
307    tokio::fs::OpenOptions::new()
308        .write(true)
309        .create_new(true)
310        .open(path)
311        .await?
312        .write_all(contents.as_ref())
313        .await
314}
315
316#[derive(Debug, Clone)]
317pub struct Spanned<T> {
318    value: T,
319    span: Span,
320}
321
322impl<T> Spanned<T> {
323    pub async fn new<F: Future<Output = T>>(span: Span, make: F) -> Self {
324        Self::try_new::<Infallible, _>(span, async { Ok(make.await) })
325            .await
326            .unwrap()
327    }
328
329    pub async fn try_new<E, F: Future<Output = Result<T, E>>>(
330        span: Span,
331        make: F,
332    ) -> Result<Self, E> {
333        let span2 = span.clone();
334        async {
335            Ok(Self {
336                value: make.await?,
337                span: span2,
338            })
339        }
340        .instrument(span)
341        .await
342    }
343
344    pub fn borrow(&self) -> Spanned<&T> {
345        Spanned {
346            value: &self.value,
347            span: self.span.clone(),
348        }
349    }
350
351    pub fn map<U>(self, map: impl Fn(T) -> U) -> Spanned<U> {
352        Spanned {
353            value: map(self.value),
354            span: self.span,
355        }
356    }
357
358    pub fn borrow_mut(&mut self) -> Spanned<&mut T> {
359        Spanned {
360            value: &mut self.value,
361            span: self.span.clone(),
362        }
363    }
364
365    pub fn with_sync<O, F: FnOnce(T) -> O>(self, f: F) -> O {
366        let _g = self.span.enter();
367        f(self.value)
368    }
369
370    pub async fn with<Fut: Future, F: FnOnce(T) -> Fut>(self, f: F) -> Fut::Output {
371        async { f(self.value).await }.instrument(self.span).await
372    }
373
374    pub fn span(&self) -> Span {
375        self.span.clone()
376    }
377
378    pub fn value(&self) -> &T {
379        &self.value
380    }
381
382    pub fn value_mut(&mut self) -> &mut T {
383        &mut self.value
384    }
385
386    pub fn into_value(self) -> T {
387        self.value
388    }
389}
390
391/// For CLIs, detects `version-hash` as a single argument, prints the provided
392/// version hash, then exits the process.
393pub fn handle_version_hash_command(version_hash: &str) {
394    let mut args = std::env::args();
395    if let Some(ref arg) = args.nth(1)
396        && arg.as_str() == "version-hash"
397    {
398        println!("{version_hash}");
399        std::process::exit(0);
400    }
401}
402
403/// Run the supplied closure `op_fn` until it succeeds. Frequency and number of
404/// retries is determined by the specified strategy.
405///
406/// ```
407/// use std::time::Duration;
408///
409/// use fedimint_core::util::{backoff_util, retry};
410/// # tokio_test::block_on(async {
411/// retry(
412///     "Gateway balance after swap".to_string(),
413///     backoff_util::background_backoff(),
414///     || async {
415///         // Fallible network calls …
416///         Ok(())
417///     },
418/// )
419/// .await
420/// .expect("never fails");
421/// # });
422/// ```
423///
424/// # Returns
425///
426/// - If the closure runs successfully, the result is immediately returned
427/// - If the closure did not run successfully for `max_attempts` times, the
428///   error of the closure is returned
429pub async fn retry<F, Fut, T>(
430    op_name: impl Into<String>,
431    strategy: impl backoff_util::Backoff,
432    op_fn: F,
433) -> Result<T, anyhow::Error>
434where
435    F: Fn() -> Fut,
436    Fut: Future<Output = Result<T, anyhow::Error>>,
437{
438    let mut strategy = strategy;
439    let op_name = op_name.into();
440    let mut attempts: u64 = 0;
441    loop {
442        attempts += 1;
443        match op_fn().await {
444            Ok(result) => return Ok(result),
445            Err(err) => {
446                if let Some(interval) = strategy.next() {
447                    // run closure op_fn again
448                    debug!(
449                        target: LOG_CORE,
450                        err = %err.fmt_compact_anyhow(),
451                        %attempts,
452                        interval = interval.as_secs(),
453                        "{} failed, retrying",
454                        op_name,
455                    );
456                    runtime::sleep(interval).await;
457                } else {
458                    warn!(
459                        target: LOG_CORE,
460                        err = %err.fmt_compact_anyhow(),
461                        %attempts,
462                        "{} failed",
463                        op_name,
464                    );
465                    return Err(err);
466                }
467            }
468        }
469    }
470}
471
472/// Computes the median from a slice of sorted `u64`s
473pub fn get_median(vals: &[u64]) -> Option<u64> {
474    if vals.is_empty() {
475        return None;
476    }
477    let len = vals.len();
478    let mid = len / 2;
479
480    if len.is_multiple_of(2) {
481        Some(u64::midpoint(vals[mid - 1], vals[mid]))
482    } else {
483        Some(vals[mid])
484    }
485}
486
487/// Computes the average of the given `u64` slice.
488pub fn get_average(vals: &[u64]) -> Option<u64> {
489    if vals.is_empty() {
490        return None;
491    }
492
493    let sum: u64 = vals.iter().sum();
494    Some(sum / vals.len() as u64)
495}
496
497#[cfg(test)]
498mod tests;
fedimint_core/util/mod.rs

fedimint_core/util/
mod.rs
