iroh_blobs/downloader/
get.rs

1//! [`Getter`] implementation that performs requests over [`Connection`]s.
2//!
3//! [`Connection`]: iroh::endpoint::Connection
4
5use futures_lite::FutureExt;
6use iroh::endpoint;
7
8use super::{progress::BroadcastProgressSender, DownloadKind, FailureAction, GetStartFut, Getter};
9use crate::{
10    get::{db::get_to_db_in_steps, error::GetError},
11    store::Store,
12};
13
14impl From<GetError> for FailureAction {
15    fn from(e: GetError) -> Self {
16        match e {
17            e @ GetError::NotFound(_) => FailureAction::AbortRequest(e.into()),
18            e @ GetError::RemoteReset(_) => FailureAction::RetryLater(e.into()),
19            e @ GetError::NoncompliantNode(_) => FailureAction::DropPeer(e.into()),
20            e @ GetError::Io(_) => FailureAction::RetryLater(e.into()),
21            e @ GetError::BadRequest(_) => FailureAction::AbortRequest(e.into()),
22            // TODO: what do we want to do on local failures?
23            e @ GetError::LocalFailure(_) => FailureAction::AbortRequest(e.into()),
24        }
25    }
26}
27
28/// [`Getter`] implementation that performs requests over [`Connection`]s.
29///
30/// [`Connection`]: iroh::endpoint::Connection
31pub(crate) struct IoGetter<S: Store> {
32    pub store: S,
33}
34
35impl<S: Store> Getter for IoGetter<S> {
36    type Connection = endpoint::Connection;
37    type NeedsConn = crate::get::db::GetStateNeedsConn;
38
39    fn get(
40        &mut self,
41        kind: DownloadKind,
42        progress_sender: BroadcastProgressSender,
43    ) -> GetStartFut<Self::NeedsConn> {
44        let store = self.store.clone();
45        async move {
46            match get_to_db_in_steps(store, kind.hash_and_format(), progress_sender).await {
47                Err(err) => Err(err.into()),
48                Ok(crate::get::db::GetState::Complete(stats)) => {
49                    Ok(super::GetOutput::Complete(stats))
50                }
51                Ok(crate::get::db::GetState::NeedsConn(needs_conn)) => {
52                    Ok(super::GetOutput::NeedsConn(needs_conn))
53                }
54            }
55        }
56        .boxed_local()
57    }
58}
59
60impl super::NeedsConn<endpoint::Connection> for crate::get::db::GetStateNeedsConn {
61    fn proceed(self, conn: endpoint::Connection) -> super::GetProceedFut {
62        async move {
63            let res = self.proceed(conn).await;
64            #[cfg(feature = "metrics")]
65            track_metrics(&res);
66            match res {
67                Ok(stats) => Ok(stats),
68                Err(err) => Err(err.into()),
69            }
70        }
71        .boxed_local()
72    }
73}
74
75#[cfg(feature = "metrics")]
76fn track_metrics(res: &Result<crate::get::Stats, GetError>) {
77    use iroh_metrics::{inc, inc_by};
78
79    use crate::metrics::Metrics;
80    match res {
81        Ok(stats) => {
82            let crate::get::Stats {
83                bytes_written,
84                bytes_read: _,
85                elapsed,
86            } = stats;
87
88            inc!(Metrics, downloads_success);
89            inc_by!(Metrics, download_bytes_total, *bytes_written);
90            inc_by!(Metrics, download_time_total, elapsed.as_millis() as u64);
91        }
92        Err(e) => match &e {
93            GetError::NotFound(_) => inc!(Metrics, downloads_notfound),
94            _ => inc!(Metrics, downloads_error),
95        },
96    }
97}
iroh_blobs/downloader/get.rs

iroh_blobs/downloader/
get.rs
