iroh_blobs/
api.rs

1//! The user facing API of the store.
2//!
3//! This API is both for interacting with an in-process store and for interacting
4//! with a remote store via rpc calls.
5//!
6//! The entry point for the api is the [`Store`] struct. There are several ways
7//! to obtain a `Store` instance: it is available via [`Deref`]
8//! from the different store implementations
9//! (e.g. [`MemStore`](crate::store::mem::MemStore)
10//! and [`FsStore`](crate::store::fs::FsStore)) as well as on the
11//! [`BlobsProtocol`](crate::BlobsProtocol) iroh protocol handler.
12//!
13//! You can also [`connect`](Store::connect) to a remote store that is listening
14//! to rpc requests.
15use std::{io, ops::Deref};
16
17use bao_tree::io::EncodeError;
18use iroh::Endpoint;
19use n0_error::{e, stack_error};
20use proto::{ShutdownRequest, SyncDbRequest};
21use ref_cast::RefCast;
22use serde::{Deserialize, Serialize};
23use tags::Tags;
24
25pub mod blobs;
26pub mod downloader;
27pub mod proto;
28pub mod remote;
29pub mod tags;
30use crate::{api::proto::WaitIdleRequest, provider::events::ProgressError};
31pub use crate::{store::util::Tag, util::temp_tag::TempTag};
32
33pub(crate) type ApiClient = irpc::Client<proto::Request>;
34
35#[allow(missing_docs)]
36#[non_exhaustive]
37#[stack_error(derive, add_meta)]
38pub enum RequestError {
39    /// Request failed due to rpc error.
40    #[error("rpc error: {source}")]
41    Rpc { source: irpc::Error },
42    /// Request failed due an actual error.
43    #[error("inner error: {source}")]
44    Inner {
45        #[error(std_err)]
46        source: Error,
47    },
48}
49
50impl From<irpc::Error> for RequestError {
51    fn from(value: irpc::Error) -> Self {
52        e!(RequestError::Rpc, value)
53    }
54}
55
56impl From<Error> for RequestError {
57    fn from(value: Error) -> Self {
58        e!(RequestError::Inner, value)
59    }
60}
61
62impl From<io::Error> for RequestError {
63    fn from(value: io::Error) -> Self {
64        e!(RequestError::Inner, value.into())
65    }
66}
67
68impl From<irpc::channel::mpsc::RecvError> for RequestError {
69    fn from(value: irpc::channel::mpsc::RecvError) -> Self {
70        e!(RequestError::Rpc, value.into())
71    }
72}
73
74pub type RequestResult<T> = std::result::Result<T, RequestError>;
75
76#[allow(missing_docs)]
77#[non_exhaustive]
78#[stack_error(derive, add_meta, from_sources)]
79pub enum ExportBaoError {
80    #[error("send error")]
81    Send { source: irpc::channel::SendError },
82    #[error("mpsc recv e api.acp.pro-channelsrror")]
83    MpscRecv {
84        source: irpc::channel::mpsc::RecvError,
85    },
86    #[error("oneshot recv error")]
87    OneshotRecv {
88        source: irpc::channel::oneshot::RecvError,
89    },
90    #[error("request error")]
91    Request { source: irpc::RequestError },
92    #[error("io error")]
93    ExportBaoIo {
94        #[error(std_err)]
95        source: io::Error,
96    },
97    #[error("encode error")]
98    ExportBaoInner {
99        #[error(std_err)]
100        source: bao_tree::io::EncodeError,
101    },
102    #[error("client error")]
103    ClientError { source: ProgressError },
104}
105
106impl From<ExportBaoError> for Error {
107    fn from(e: ExportBaoError) -> Self {
108        match e {
109            ExportBaoError::Send { source, .. } => Self::Io(source.into()),
110            ExportBaoError::MpscRecv { source, .. } => Self::Io(source.into()),
111            ExportBaoError::OneshotRecv { source, .. } => Self::Io(source.into()),
112            ExportBaoError::Request { source, .. } => Self::Io(source.into()),
113            ExportBaoError::ExportBaoIo { source, .. } => Self::Io(source),
114            ExportBaoError::ExportBaoInner { source, .. } => Self::Io(source.into()),
115            ExportBaoError::ClientError { source, .. } => Self::Io(source.into()),
116        }
117    }
118}
119
120impl From<irpc::Error> for ExportBaoError {
121    fn from(e: irpc::Error) -> Self {
122        match e {
123            irpc::Error::MpscRecv { source: e, .. } => e!(ExportBaoError::MpscRecv, e),
124            irpc::Error::OneshotRecv { source: e, .. } => e!(ExportBaoError::OneshotRecv, e),
125            irpc::Error::Send { source: e, .. } => e!(ExportBaoError::Send, e),
126            irpc::Error::Request { source: e, .. } => e!(ExportBaoError::Request, e),
127            #[cfg(feature = "rpc")]
128            irpc::Error::Write { source: e, .. } => e!(ExportBaoError::ExportBaoIo, e.into()),
129        }
130    }
131}
132
133pub type ExportBaoResult<T> = std::result::Result<T, ExportBaoError>;
134
135#[derive(Serialize, Deserialize)]
136#[stack_error(derive, std_sources, from_sources)]
137pub enum Error {
138    #[serde(with = "crate::util::serde::io_error_serde")]
139    Io(#[error(source)] io::Error),
140}
141
142impl Error {
143    pub fn io(
144        kind: io::ErrorKind,
145        msg: impl Into<Box<dyn std::error::Error + Send + Sync>>,
146    ) -> Self {
147        Self::Io(io::Error::new(kind, msg.into()))
148    }
149
150    pub fn other<E>(msg: E) -> Self
151    where
152        E: Into<Box<dyn std::error::Error + Send + Sync>>,
153    {
154        Self::Io(io::Error::other(msg.into()))
155    }
156}
157
158impl From<irpc::Error> for Error {
159    fn from(e: irpc::Error) -> Self {
160        Self::Io(e.into())
161    }
162}
163
164impl From<RequestError> for Error {
165    fn from(e: RequestError) -> Self {
166        match e {
167            RequestError::Rpc { source, .. } => Self::Io(source.into()),
168            RequestError::Inner { source, .. } => source,
169        }
170    }
171}
172
173impl From<irpc::channel::mpsc::RecvError> for Error {
174    fn from(e: irpc::channel::mpsc::RecvError) -> Self {
175        Self::Io(e.into())
176    }
177}
178
179#[cfg(feature = "rpc")]
180#[cfg_attr(iroh_blobs_docsrs, doc(cfg(feature = "rpc")))]
181impl From<irpc::rpc::WriteError> for Error {
182    fn from(e: irpc::rpc::WriteError) -> Self {
183        Self::Io(e.into())
184    }
185}
186
187impl From<irpc::RequestError> for Error {
188    fn from(e: irpc::RequestError) -> Self {
189        Self::Io(e.into())
190    }
191}
192
193impl From<irpc::channel::SendError> for Error {
194    fn from(e: irpc::channel::SendError) -> Self {
195        Self::Io(e.into())
196    }
197}
198
199impl From<EncodeError> for Error {
200    fn from(value: EncodeError) -> Self {
201        match value {
202            EncodeError::Io(cause) => Self::Io(cause),
203            _ => Self::Io(io::Error::other(value)),
204        }
205    }
206}
207
208pub type Result<T> = std::result::Result<T, Error>;
209
210/// The main entry point for the store API.
211#[derive(Debug, Clone, ref_cast::RefCast)]
212#[repr(transparent)]
213pub struct Store {
214    client: ApiClient,
215}
216
217impl Deref for Store {
218    type Target = blobs::Blobs;
219
220    fn deref(&self) -> &Self::Target {
221        blobs::Blobs::ref_from_sender(&self.client)
222    }
223}
224
225impl Store {
226    /// The tags API.
227    pub fn tags(&self) -> &Tags {
228        Tags::ref_from_sender(&self.client)
229    }
230
231    /// The blobs API.
232    pub fn blobs(&self) -> &blobs::Blobs {
233        blobs::Blobs::ref_from_sender(&self.client)
234    }
235
236    /// API for getting blobs from a *single* remote node.
237    pub fn remote(&self) -> &remote::Remote {
238        remote::Remote::ref_from_sender(&self.client)
239    }
240
241    /// Create a downloader for more complex downloads.
242    ///
243    /// Unlike the other APIs, this creates an object that has internal state,
244    /// so don't create it ad hoc but store it somewhere if you need it multiple
245    /// times.
246    pub fn downloader(&self, endpoint: &Endpoint) -> downloader::Downloader {
247        downloader::Downloader::new(self, endpoint)
248    }
249
250    /// Connect to a remote store as a rpc client.
251    #[cfg(feature = "rpc")]
252    #[cfg_attr(iroh_blobs_docsrs, doc(cfg(feature = "rpc")))]
253    pub fn connect(endpoint: noq::Endpoint, addr: std::net::SocketAddr) -> Self {
254        let sender = irpc::Client::noq(endpoint, addr);
255        Store::from_sender(sender)
256    }
257
258    /// Listen on a noq endpoint for incoming rpc connections.
259    #[cfg(feature = "rpc")]
260    #[cfg_attr(iroh_blobs_docsrs, doc(cfg(feature = "rpc")))]
261    pub async fn listen(self, endpoint: noq::Endpoint) {
262        use irpc::rpc::RemoteService;
263
264        use self::proto::Request;
265        let local = self.client.as_local().unwrap().clone();
266        let handler = Request::remote_handler(local);
267        irpc::rpc::listen::<Request>(endpoint, handler).await
268    }
269
270    pub async fn sync_db(&self) -> RequestResult<()> {
271        let msg = SyncDbRequest;
272        self.client.rpc(msg).await??;
273        Ok(())
274    }
275
276    pub async fn shutdown(&self) -> irpc::Result<()> {
277        let msg = ShutdownRequest;
278        self.client.rpc(msg).await?;
279        Ok(())
280    }
281
282    /// Waits for the store to become completely idle.
283    ///
284    /// This is mostly useful for tests, where you want to check that e.g. the
285    /// store has written all data to disk.
286    ///
287    /// Note that a store is not guaranteed to become idle, if it is being
288    /// interacted with concurrently. So this might wait forever.
289    ///
290    /// Also note that once you get the callback, the store is not guaranteed to
291    /// still be idle. All this tells you that there was a point in time where
292    /// the store was idle between the call and the response.
293    pub async fn wait_idle(&self) -> irpc::Result<()> {
294        let msg = WaitIdleRequest;
295        self.client.rpc(msg).await?;
296        Ok(())
297    }
298
299    pub(crate) fn from_sender(client: ApiClient) -> Self {
300        Self { client }
301    }
302
303    pub(crate) fn ref_from_sender(client: &ApiClient) -> &Self {
304        Self::ref_cast(client)
305    }
306}
iroh_blobs/api.rs

iroh_blobs/
api.rs
