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

fedimint_core/util/
mod.rs
