syncweb_core/node/
blob_store.rs

1use bytes::Bytes;
2use iroh::{EndpointAddr, address_lookup::memory::MemoryLookup, endpoint::Endpoint};
3use iroh_blobs::{
4    BlobFormat, BlobsProtocol, Hash,
5    api::{
6        Store as BlobApi,
7        blobs::{AddPathOptions, BlobReader, BlobStatus, ExportMode, ExportOptions, ImportMode},
8    },
9    protocol::GetRequest,
10    ticket::BlobTicket,
11};
12use n0_future::StreamExt;
13use std::path::{Path, PathBuf};
14
15use crate::error::{Result, SyncwebError};
16
17#[derive(Clone)]
18pub struct BlobStore {
19    store: BlobApi,
20    address_lookup: MemoryLookup,
21}
22
23impl BlobStore {
24    #[must_use]
25    pub fn new(protocol: &BlobsProtocol) -> Self {
26        Self::new_with_address_lookup(protocol, MemoryLookup::new())
27    }
28
29    #[must_use]
30    pub fn new_with_address_lookup(protocol: &BlobsProtocol, address_lookup: MemoryLookup) -> Self {
31        Self {
32            store: protocol.store().clone(),
33            address_lookup,
34        }
35    }
36
37    #[must_use]
38    pub const fn inner(&self) -> &BlobApi {
39        &self.store
40    }
41
42    /// Register a peer's known addresses so later connections (such as blob
43    /// downloads that only carry the peer's node id) can resolve the peer
44    /// without LAN discovery.
45    pub fn add_endpoint_addr(&self, addr: EndpointAddr) {
46        self.address_lookup.add_endpoint_info(addr);
47    }
48
49    /// Create a reader for the given hash.
50    ///
51    /// The reader implements [`tokio::io::AsyncRead`] and [`tokio::io::AsyncSeek`],
52    /// allowing O(1)-memory streaming to any sink.
53    #[must_use]
54    pub fn reader(&self, hash: impl Into<Hash>) -> BlobReader {
55        self.store.reader(hash)
56    }
57
58    /// # Errors
59    ///
60    /// Returns an error if the blob fails to be added to the store.
61    pub async fn add_bytes(&self, data: impl AsRef<[u8]>) -> Result<Hash> {
62        Ok(self
63            .store
64            .add_bytes(Bytes::copy_from_slice(data.as_ref()))
65            .await
66            .map_err(|error| SyncwebError::operation("failed to add blob bytes", error))?
67            .hash)
68    }
69
70    /// # Errors
71    ///
72    /// Returns an error if the file fails to be read or added to the store.
73    pub async fn add_file(&self, path: impl AsRef<Path>) -> Result<Hash> {
74        let absolute_path = absolute_path(path)?;
75        Ok(self
76            .store
77            .add_path(absolute_path)
78            .await
79            .map_err(|error| SyncwebError::operation("failed to add blob file", error))?
80            .hash)
81    }
82
83    /// Add a file using a reference to its original path instead of copying
84    /// its contents into the blob store.
85    ///
86    /// The source file must remain available and unchanged for as long as the
87    /// blob store may need to read this blob.
88    ///
89    /// # Errors
90    ///
91    /// Returns an error if the file fails to be read or added to the store.
92    pub async fn add_file_ref(&self, path: impl AsRef<Path>) -> Result<Hash> {
93        let absolute_path = absolute_path(path)?;
94        Ok(self
95            .store
96            .add_path_with_opts(AddPathOptions {
97                path: absolute_path,
98                mode: ImportMode::TryReference,
99                format: BlobFormat::Raw,
100            })
101            .await
102            .map_err(|error| SyncwebError::operation("failed to add blob file (reference)", error))?
103            .hash)
104    }
105
106    /// Export a complete blob to a temporary or caller-managed path.
107    ///
108    /// Uses [`ExportMode::Copy`] — the blob stays in the store and a copy
109    /// (or `CoW` reflink) is written to `destination`.
110    ///
111    /// # Errors
112    ///
113    /// Returns an error if the blob cannot be read or written.
114    pub async fn export_to_path(&self, hash: Hash, destination: impl AsRef<Path>) -> Result<u64> {
115        self.export_to_path_with_mode(hash, destination, ExportMode::Copy).await
116    }
117
118    /// Export a complete blob with a specific [`ExportMode`].
119    ///
120    /// `ExportMode::TryReference` moves the blob's internal file to
121    /// `destination` and updates the store to track it externally,
122    /// avoiding a copy. `ExportMode::Copy` leaves the blob in the store
123    /// and writes a copy (or `CoW` reflink) to `destination`.
124    ///
125    /// # Errors
126    ///
127    /// Returns an error if the blob cannot be read or written.
128    pub async fn export_to_path_with_mode(
129        &self,
130        hash: Hash,
131        destination: impl AsRef<Path>,
132        mode: ExportMode,
133    ) -> Result<u64> {
134        self.store
135            .export_with_opts(ExportOptions {
136                hash,
137                mode,
138                target: destination.as_ref().to_owned(),
139            })
140            .await
141            .map_err(|error| SyncwebError::operation("failed to export blob", crate::error::error_chain(&error)))
142    }
143
144    /// # Errors
145    ///
146    /// Returns an error if the store cannot be queried.
147    pub async fn has(&self, hash: Hash) -> Result<bool> {
148        self.store
149            .has(hash)
150            .await
151            .map_err(|error| SyncwebError::operation("failed to query blob store", error))
152    }
153
154    /// Get metadata (size) for a blob.
155    ///
156    /// # Errors
157    ///
158    /// Returns an error if the blob is not found or the store cannot be queried.
159    pub async fn stat(&self, hash: Hash) -> Result<BlobStatus> {
160        self.store
161            .blobs()
162            .status(hash)
163            .await
164            .map_err(|error| SyncwebError::operation("failed to stat blob", error))
165    }
166
167    /// # Errors
168    ///
169    /// Returns an error if the blob cannot be found or read.
170    pub async fn get(&self, hash: Hash) -> Result<Bytes> {
171        self.store
172            .get_bytes(hash)
173            .await
174            .map_err(|error| SyncwebError::operation("failed to read blob", error))
175    }
176
177    #[must_use]
178    pub fn ticket(&self, endpoint: &Endpoint, hash: Hash) -> BlobTicket {
179        self.ticket_for_addr(endpoint.addr(), hash)
180    }
181
182    #[must_use]
183    pub fn ticket_for_addr(&self, addr: EndpointAddr, hash: Hash) -> BlobTicket {
184        raw_blob_ticket(addr, hash)
185    }
186
187    /// Pin a blob with a durable named tag so garbage collection cannot remove it.
188    ///
189    /// # Errors
190    ///
191    /// Returns an error if the tag cannot be written.
192    pub async fn pin(&self, name: impl AsRef<str>, hash: Hash) -> Result<()> {
193        self.store
194            .tags()
195            .set(name.as_ref(), hash)
196            .await
197            .map_err(|error| SyncwebError::operation("failed to pin blob", error))
198    }
199
200    /// Remove a durable pin from a blob.
201    ///
202    /// # Errors
203    ///
204    /// Returns an error if the tag cannot be removed.
205    pub async fn unpin(&self, name: impl AsRef<str>) -> Result<()> {
206        self.store
207            .tags()
208            .delete(name.as_ref())
209            .await
210            .map(|_deleted| ())
211            .map_err(|error| SyncwebError::operation("failed to unpin blob", error))
212    }
213
214    /// # Errors
215    ///
216    /// Returns an error if the pin cannot be queried.
217    pub async fn is_pinned(&self, name: impl AsRef<str>, hash: Hash) -> Result<bool> {
218        self.store
219            .tags()
220            .get(name.as_ref())
221            .await
222            .map(|tag_info| tag_info.is_some_and(|tag| tag.hash == hash))
223            .map_err(|error| SyncwebError::operation("failed to query blob pin", error))
224    }
225
226    /// List all complete blobs currently in the store.
227    ///
228    /// # Errors
229    ///
230    /// Returns an error if the store cannot be queried.
231    pub async fn list_hashes(&self) -> Result<Vec<Hash>> {
232        self.store
233            .blobs()
234            .list()
235            .hashes()
236            .await
237            .map_err(|error| SyncwebError::operation("failed to list blobs", error))
238    }
239
240    /// List named pins whose names start with `prefix`.
241    ///
242    /// # Errors
243    ///
244    /// Returns an error if the store cannot be queried.
245    pub async fn list_pins(&self, prefix: impl AsRef<[u8]>) -> Result<Vec<(String, Hash)>> {
246        let mut tags = self
247            .store
248            .tags()
249            .list_prefix(prefix)
250            .await
251            .map_err(|error| SyncwebError::operation("failed to list blob pins", error))?;
252        let mut pins = Vec::new();
253        while let Some(tag_result) = tags.next().await {
254            let tag = tag_result.map_err(|error| SyncwebError::operation("failed to read blob pin", error))?;
255            let name = String::from_utf8(tag.name.as_ref().to_vec())
256                .map_err(|error| SyncwebError::operation("blob pin name is not UTF-8", error))?;
257            pins.push((name, tag.hash));
258        }
259        Ok(pins)
260    }
261
262    /// # Errors
263    ///
264    /// Returns an error if the blob cannot be fetched from the remote endpoint.
265    pub async fn fetch(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<()> {
266        self.address_lookup.add_endpoint_info(ticket.addr().clone());
267        self.store
268            .downloader(endpoint)
269            .download(ticket.hash_and_format(), [ticket.addr().id])
270            .await
271            .map_err(|error| SyncwebError::operation("failed to fetch blob", error))?;
272        Ok(())
273    }
274
275    /// Fetch a blob through the first working provider ticket and pin it,
276    /// skipping the fetch when the blob is already present locally.
277    ///
278    /// # Errors
279    ///
280    /// Returns an error if the blob cannot be fetched through any ticket.
281    pub async fn fetch_and_pin(&self, endpoint: &Endpoint, content_hash: Hash, tickets: &[BlobTicket]) -> Result<()> {
282        if !self.has(content_hash).await? {
283            let candidates: Vec<BlobTicket> = tickets
284                .iter()
285                .filter(|ticket| ticket.hash() == content_hash)
286                .cloned()
287                .collect();
288            try_fetch_first_working(&candidates, |ticket| self.fetch(endpoint, ticket)).await?;
289        }
290        let pin_name = crate::pins::download_pin(content_hash);
291        self.pin(&pin_name, content_hash).await?;
292        Ok(())
293    }
294
295    /// Force-fetch a blob from a remote peer, re-downloading even if the blob
296    /// is already present locally. Uses `iroh-blobs`' low-level `execute_get`
297    /// API which does not check local availability.
298    ///
299    /// # Errors
300    ///
301    /// Returns an error if the connection or download fails.
302    pub async fn force_fetch(&self, endpoint: &Endpoint, ticket: &BlobTicket) -> Result<()> {
303        self.address_lookup.add_endpoint_info(ticket.addr().clone());
304        let hash = ticket.hash();
305        let conn = endpoint
306            .connect(ticket.addr().id, iroh_blobs::protocol::ALPN)
307            .await
308            .map_err(|error| SyncwebError::operation("failed to connect to provider", error))?;
309        self.force_fetch_from_conn(conn, hash).await
310    }
311
312    /// Force-fetch a blob from a known peer, connecting by its discovered
313    /// address (which may include a relay or custom transport address).
314    ///
315    /// # Errors
316    ///
317    /// Returns an error if the connection or download fails.
318    pub async fn force_fetch_from_peer(
319        &self,
320        endpoint: &Endpoint,
321        peer: &iroh::EndpointAddr,
322        hash: Hash,
323    ) -> Result<()> {
324        self.address_lookup.add_endpoint_info(peer.clone());
325        let conn = endpoint
326            .connect(peer.id, iroh_blobs::protocol::ALPN)
327            .await
328            .map_err(|error| SyncwebError::operation("failed to connect to peer", error))?;
329        self.force_fetch_from_conn(conn, hash).await
330    }
331
332    async fn force_fetch_from_conn(&self, conn: iroh::endpoint::Connection, hash: Hash) -> Result<()> {
333        let request = GetRequest::blob(hash);
334        self.store
335            .remote()
336            .execute_get(conn, request)
337            .await
338            .map_err(|error| SyncwebError::operation("failed to force-fetch blob", error))?;
339        Ok(())
340    }
341}
342
343fn absolute_path(path: impl AsRef<Path>) -> Result<PathBuf> {
344    std::path::absolute(path.as_ref())
345        .map_err(|error| SyncwebError::operation("failed to resolve blob file path", error))
346}
347
348/// Build an unauthenticated blob ticket for a content hash hosted by an address.
349#[must_use]
350pub fn raw_blob_ticket(addr: EndpointAddr, hash: Hash) -> BlobTicket {
351    BlobTicket::new(addr, hash, BlobFormat::Raw)
352}
353
354/// Try each provider in order until one succeeds, returning the first success.
355/// This is the shared "first working provider wins" primitive used by
356/// fetch-with-fallback flows; [`crate::indexing::links::fetch_from_mirrors`] is
357/// the synchronous twin for resolver closures.
358///
359/// # Errors
360///
361/// Returns the last fetch error, or a generic error when no provider succeeded.
362pub(crate) async fn try_fetch_first_working<'a, T, U, F, Fut>(items: &'a [T], mut fetch: F) -> Result<U>
363where
364    T: Sync,
365    F: FnMut(&'a T) -> Fut + Send,
366    Fut: std::future::Future<Output = Result<U>> + Send + 'a,
367{
368    let mut last_error: Option<SyncwebError> = None;
369    for item in items {
370        match fetch(item).await {
371            Ok(value) => return Ok(value),
372            Err(error) => last_error = Some(error),
373        }
374    }
375    Err(last_error.unwrap_or_else(|| SyncwebError::operation("failed to fetch blob", "no working provider ticket")))
376}
syncweb_core/node/blob_store.rs

syncweb_core/node/
blob_store.rs
