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

fedimint_core/util/
mod.rs
