rust_ipfs/repo/
mod.rs

1//! Storage implementation(s) backing the [`crate::Ipfs`].
2use crate::Block;
3use crate::error::Error;
4use connexa::prelude::identity::PeerId;
5use core::fmt::Debug;
6use futures::channel::mpsc::{Receiver, Sender, channel};
7use futures::future::{BoxFuture, Either};
8use futures::sink::SinkExt;
9use futures::stream::{self, BoxStream, FuturesOrdered, FuturesUnordered};
10use futures::{FutureExt, Stream, StreamExt, TryStreamExt};
11use indexmap::IndexSet;
12use ipld_core::cid::Cid;
13use parking_lot::{Mutex, RwLock};
14use std::borrow::Borrow;
15use std::collections::{BTreeSet, HashMap, HashSet};
16use std::future::{Future, IntoFuture};
17#[allow(unused_imports)]
18use std::path::Path;
19use std::pin::Pin;
20use std::sync::Arc;
21use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
22use std::task::{Context, Poll};
23use std::time::Duration;
24use std::{error, fmt, io};
25use tokio::sync::RwLockReadGuard;
26use tracing::{Instrument, Span};
27
28#[macro_use]
29#[cfg(test)]
30mod common_tests;
31
32pub use store::{
33    blockstore, datastore, default_impl::DefaultStorage, default_impl::keystore::DefaultKeystore,
34};
35
36pub mod lock;
37
38/// Path mangling done for pins and blocks
39#[cfg(not(target_arch = "wasm32"))]
40pub(crate) mod paths;
41mod store;
42
43/// Describes the outcome of `BlockStore::put_block`.
44#[derive(Debug, PartialEq, Eq)]
45pub enum BlockPut {
46    /// A new block was written to the blockstore.
47    NewBlock,
48    /// The block already exists.
49    Existed,
50}
51
52/// Describes the outcome of `BlockStore::remove`.
53#[derive(Debug)]
54pub enum BlockRm {
55    /// A block was successfully removed from the blockstore.
56    Removed(Cid),
57    // TODO: DownloadCancelled(Cid, Duration),
58}
59
60// pub struct BlockNotFound(Cid);
61/// Describes the error variants for `BlockStore::remove`.
62#[derive(Debug)]
63pub enum BlockRmError {
64    // TODO: Pinned(Cid),
65    /// The `Cid` doesn't correspond to a block in the blockstore.
66    NotFound(Cid),
67}
68
69pub trait StoreOpt {
70    fn into_opt(self) -> Self;
71}
72
73/// This API is being discussed and evolved, which will likely lead to breakage.
74pub trait BlockStore: Debug + Send + Sync {
75    fn init(&self) -> impl Future<Output = Result<(), Error>> + Send;
76
77    /// Returns whether a block is present in the blockstore.
78    fn contains(&self, cid: &Cid) -> impl Future<Output = Result<bool, Error>> + Send;
79    /// Returns a block from the blockstore.
80    fn get(&self, cid: &Cid) -> impl Future<Output = Result<Option<Block>, Error>> + Send;
81    /// Get the size of a single block
82    fn size(&self, cid: &[Cid]) -> impl Future<Output = Result<Option<usize>, Error>> + Send;
83    /// Get a total size of the block store
84    fn total_size(&self) -> impl Future<Output = Result<usize, Error>> + Send;
85    /// Inserts a block in the blockstore.
86    fn put(&self, block: &Block) -> impl Future<Output = Result<(Cid, BlockPut), Error>> + Send;
87    /// Inserts multiple blocks in one batch.
88    fn put_many(
89        &self,
90        blocks: &[Block],
91    ) -> impl Future<Output = Result<Vec<(Cid, BlockPut)>, Error>> + Send {
92        async move {
93            let mut out = Vec::with_capacity(blocks.len());
94            for block in blocks {
95                out.push(self.put(block).await?);
96            }
97            Ok(out)
98        }
99    }
100    /// Removes a block from the blockstore.
101    fn remove(&self, cid: &Cid) -> impl Future<Output = Result<(), Error>> + Send;
102    /// Remove multiple blocks from the blockstore
103    fn remove_many(
104        &self,
105        blocks: BoxStream<'static, Cid>,
106    ) -> impl Future<Output = BoxStream<'static, Cid>> + Send;
107    /// Returns a list of the blocks (Cids), in the blockstore.
108    fn list(&self) -> impl Future<Output = BoxStream<'static, Cid>> + Send;
109}
110
111/// Generic layer of abstraction for a key-value data store.
112pub trait DataStore: PinStore + Debug + Send + Sync {
113    fn init(&self) -> impl Future<Output = Result<(), Error>> + Send;
114    /// Checks if a key is present in the datastore.
115    fn contains(&self, key: &[u8]) -> impl Future<Output = Result<bool, Error>> + Send;
116    /// Returns the value associated with a key from the datastore.
117    fn get(&self, key: &[u8]) -> impl Future<Output = Result<Option<Vec<u8>>, Error>> + Send;
118    /// Puts the value under the key in the datastore.
119    fn put(&self, key: &[u8], value: &[u8]) -> impl Future<Output = Result<(), Error>> + Send;
120    /// Removes a key-value pair from the datastore.
121    fn remove(&self, key: &[u8]) -> impl Future<Output = Result<(), Error>> + Send;
122    /// Iterate over the k/v of the datastore
123    fn iter(
124        &self,
125    ) -> impl Future<Output = futures::stream::BoxStream<'static, (Vec<u8>, Vec<u8>)>> + Send;
126}
127
128#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord)]
129pub struct GCConfig {
130    /// How long until GC runs
131    /// If duration is not set, it will not run at a timer
132    pub duration: Duration,
133
134    /// What will trigger GC
135    pub trigger: GCTrigger,
136}
137
138impl Default for GCConfig {
139    fn default() -> Self {
140        Self {
141            duration: Duration::from_secs(60 * 60),
142            trigger: GCTrigger::default(),
143        }
144    }
145}
146
147#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, PartialOrd, Ord)]
148pub enum GCTrigger {
149    /// At a specific size. If the size is at or exceeds, it will trigger GC
150    At {
151        size: usize,
152    },
153
154    AtStorage,
155
156    /// No trigger
157    #[default]
158    None,
159}
160
161/// Errors variants describing the possible failures for `Lock::try_exclusive`.
162#[derive(Debug)]
163pub enum LockError {
164    RepoInUse,
165    LockFileOpenFailed(io::Error),
166}
167
168impl fmt::Display for LockError {
169    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
170        let msg = match self {
171            LockError::RepoInUse => "The repository is already being used by an IPFS instance.",
172            LockError::LockFileOpenFailed(_) => "Failed to open repository lock file.",
173        };
174
175        write!(f, "{msg}")
176    }
177}
178
179impl From<io::Error> for LockError {
180    fn from(error: io::Error) -> Self {
181        match error.kind() {
182            // `WouldBlock` is not used by `OpenOptions` (this could change), and can therefore be
183            // matched on for the fs2 error in `FsLock::try_exclusive`.
184            io::ErrorKind::WouldBlock => LockError::RepoInUse,
185            _ => LockError::LockFileOpenFailed(error),
186        }
187    }
188}
189
190impl error::Error for LockError {
191    fn source(&self) -> Option<&(dyn error::Error + 'static)> {
192        if let Self::LockFileOpenFailed(error) = self {
193            Some(error)
194        } else {
195            None
196        }
197    }
198}
199
200/// A trait for describing repository locking.
201///
202/// This ensures no two IPFS nodes can be started with the same peer ID, as exclusive access to the
203/// repository is guaranteed. This is most useful when using a fs backed repo.
204pub trait Lock: Debug + Send + Sync {
205    // fn new(path: PathBuf) -> Self;
206    fn try_exclusive(&self) -> Result<(), LockError>;
207}
208
209type References<'a> = BoxStream<'a, Result<Cid, crate::refs::IpldRefsError>>;
210
211pub trait PinStore: Debug + Send + Sync {
212    fn is_pinned(&self, block: &Cid) -> impl Future<Output = Result<bool, Error>> + Send;
213
214    fn insert_direct_pin(&self, target: &Cid) -> impl Future<Output = Result<(), Error>> + Send;
215
216    fn insert_recursive_pin(
217        &self,
218        target: &Cid,
219        referenced: References<'_>,
220    ) -> impl Future<Output = Result<(), Error>> + Send;
221
222    fn remove_direct_pin(&self, target: &Cid) -> impl Future<Output = Result<(), Error>> + Send;
223
224    fn remove_recursive_pin(
225        &self,
226        target: &Cid,
227        referenced: References<'_>,
228    ) -> impl Future<Output = Result<(), Error>> + Send;
229
230    fn list(
231        &self,
232        mode: Option<PinMode>,
233    ) -> impl Future<Output = futures::stream::BoxStream<'static, Result<(Cid, PinMode), Error>>> + Send;
234
235    /// Returns error if any of the ids isn't pinned in the required type, otherwise returns
236    /// the pin details if all of the cids are pinned in one way or the another.
237    fn query(
238        &self,
239        ids: Vec<Cid>,
240        requirement: Option<PinMode>,
241    ) -> impl Future<Output = Result<Vec<(Cid, PinKind<Cid>)>, Error>> + Send;
242}
243
244/// `PinMode` is the description of pin type for quering purposes.
245#[derive(Debug, PartialEq, Eq, Clone, Copy)]
246pub enum PinMode {
247    Indirect,
248    Direct,
249    Recursive,
250}
251
252/// Helper for around the quite confusing test required in [`PinStore::list`] and
253/// [`PinStore::query`].
254#[derive(Debug, Clone, Copy)]
255enum PinModeRequirement {
256    Only(PinMode),
257    Any,
258}
259
260impl From<Option<PinMode>> for PinModeRequirement {
261    fn from(filter: Option<PinMode>) -> Self {
262        match filter {
263            Some(one) => PinModeRequirement::Only(one),
264            None => PinModeRequirement::Any,
265        }
266    }
267}
268
269#[allow(dead_code)]
270impl PinModeRequirement {
271    fn is_indirect_or_any(&self) -> bool {
272        use PinModeRequirement::*;
273        match self {
274            Only(PinMode::Indirect) | Any => true,
275            Only(_) => false,
276        }
277    }
278
279    fn matches<P: PartialEq<PinMode>>(&self, other: &P) -> bool {
280        use PinModeRequirement::*;
281        match self {
282            Only(one) if other == one => true,
283            Only(_) => false,
284            Any => true,
285        }
286    }
287
288    fn required(&self) -> Option<PinMode> {
289        use PinModeRequirement::*;
290        match self {
291            Only(one) => Some(*one),
292            Any => None,
293        }
294    }
295}
296
297impl<B: Borrow<Cid>> PartialEq<PinMode> for PinKind<B> {
298    fn eq(&self, other: &PinMode) -> bool {
299        matches!(
300            (self, other),
301            (PinKind::IndirectFrom(_), PinMode::Indirect)
302                | (PinKind::Direct, PinMode::Direct)
303                | (PinKind::Recursive(_), PinMode::Recursive)
304                | (PinKind::RecursiveIntention, PinMode::Recursive)
305        )
306    }
307}
308
309/// `PinKind` is more specific pin description for writing purposes. Implements
310/// `PartialEq<&PinMode>`. Generic over `Borrow<Cid>` to allow storing both reference and owned
311/// value of Cid.
312#[derive(Debug, PartialEq, Eq)]
313pub enum PinKind<C: Borrow<Cid>> {
314    IndirectFrom(C),
315    Direct,
316    Recursive(u64),
317    RecursiveIntention,
318}
319
320impl<C: Borrow<Cid>> PinKind<C> {
321    fn as_ref(&self) -> PinKind<&'_ Cid> {
322        match self {
323            PinKind::IndirectFrom(c) => PinKind::IndirectFrom(c.borrow()),
324            PinKind::Direct => PinKind::Direct,
325            PinKind::Recursive(count) => PinKind::Recursive(*count),
326            PinKind::RecursiveIntention => PinKind::RecursiveIntention,
327        }
328    }
329}
330
331type SubscriptionsMap =
332    HashMap<Cid, HashMap<u64, futures::channel::oneshot::Sender<Result<Block, String>>>>;
333
334static SUBSCRIPTION_TOKEN: AtomicU64 = AtomicU64::new(0);
335
336struct WaiterGuard<S: RepoTypes> {
337    repo: Repo<S>,
338    events: Sender<RepoEvent>,
339    cid: Cid,
340    token: u64,
341}
342
343impl<S: RepoTypes> Drop for WaiterGuard<S> {
344    fn drop(&mut self) {
345        {
346            let mut map = self.repo.inner.subscriptions.lock();
347            if let Some(inner) = map.get_mut(&self.cid) {
348                inner.remove(&self.token);
349                if inner.is_empty() {
350                    map.remove(&self.cid);
351                }
352            }
353        }
354        let _ = self.events.try_send(RepoEvent::UnwantBlock(self.cid));
355    }
356}
357
358/// Represents the configuration of the Ipfs node, its backing blockstore and datastore.
359pub trait StorageTypes: RepoTypes {}
360impl<T: RepoTypes> StorageTypes for T {}
361
362pub trait RepoTypes: Clone + Send + Sync + 'static {
363    /// Describes a blockstore.
364    type TBlockStore: BlockStore;
365    /// Describes a datastore.
366    type TDataStore: DataStore;
367    type TLock: Lock;
368}
369
370/// Describes a repo.
371/// Consolidates a blockstore, a datastore and a subscription registry.
372#[derive(Debug, Clone)]
373pub struct Repo<S: RepoTypes> {
374    pub(crate) inner: Arc<RepoInner<S>>,
375}
376
377#[derive(Debug)]
378pub(crate) struct RepoInner<S: RepoTypes> {
379    online: AtomicBool,
380    initialized: AtomicBool,
381    max_storage_size: AtomicUsize,
382    block_store: S::TBlockStore,
383    data_store: S::TDataStore,
384    events: RwLock<Option<Sender<RepoEvent>>>,
385    pub(crate) subscriptions: Mutex<SubscriptionsMap>,
386    lockfile: S::TLock,
387    pub(crate) gclock: tokio::sync::RwLock<()>,
388    pub(crate) mfs_root: tokio::sync::Mutex<(bool, Option<Cid>)>,
389}
390
391/// Events used to communicate to the swarm on repo changes.
392#[derive(Debug)]
393pub enum RepoEvent {
394    /// Signals desired blocks.
395    WantBlock(Vec<Cid>, Vec<PeerId>, Option<Duration>),
396    /// Signals that a waiter for the cid went away; the cid should be re-evaluated for cancellation
397    /// against the (authoritative) subscriptions map.
398    UnwantBlock(Cid),
399    /// Signals the posession of a new block.
400    NewBlock(Block),
401    /// Signals the removal of a block.
402    RemovedBlock(Cid),
403}
404
405impl Repo<DefaultStorage> {
406    // pub fn new(repo_type: &mut StorageType) -> Self {
407    //     match repo_type {
408    //         StorageType::Memory => Repo::new_memory(),
409    //         #[cfg(not(target_arch = "wasm32"))]
410    //         StorageType::Disk(path) => Repo::new_fs(path),
411    //         #[cfg(target_arch = "wasm32")]
412    //         StorageType::IndexedDb { namespace } => Repo::new_idb(namespace.take()),
413    //         StorageType::Custom {
414    //             blockstore,
415    //             datastore,
416    //             lock,
417    //         } => Repo::new_raw(
418    //             blockstore.take().expect("Requires blockstore"),
419    //             datastore.take().expect("Requires datastore"),
420    //             lock.take()
421    //                 .expect("Requires lockfile for data and block store"),
422    //         ),
423    //     }
424    // }
425
426    #[cfg(not(target_arch = "wasm32"))]
427    pub fn new_fs(path: impl AsRef<Path>) -> Self {
428        let mut default = DefaultStorage::default();
429
430        let path = path.as_ref().to_path_buf();
431        let mut blockstore_path = path.clone();
432        let mut datastore_path = path.clone();
433        let mut lockfile_path = path;
434        blockstore_path.push("blockstore");
435        datastore_path.push("datastore");
436        lockfile_path.push("repo_lock");
437
438        default.set_blockstore_path(blockstore_path);
439        default.set_datastore_path(datastore_path);
440        default.set_lockfile(lockfile_path);
441
442        Self::new_raw(default.clone(), default.clone(), default)
443    }
444
445    pub fn new_memory() -> Self {
446        let default = DefaultStorage::default();
447        Self::new_raw(default.clone(), default.clone(), default)
448    }
449
450    #[cfg(target_arch = "wasm32")]
451    pub fn new_idb(namespace: Option<String>) -> Self {
452        let mut default = DefaultStorage::default();
453        default.set_namespace(namespace);
454        Self::new_raw(default.clone(), default.clone(), default)
455    }
456}
457
458impl<S: RepoTypes> Repo<S> {
459    pub fn new_raw(
460        block_store: S::TBlockStore,
461        data_store: S::TDataStore,
462        lockfile: S::TLock,
463    ) -> Self {
464        let inner = RepoInner {
465            initialized: AtomicBool::default(),
466            online: AtomicBool::default(),
467            block_store,
468            data_store,
469            events: Default::default(),
470            subscriptions: Default::default(),
471            lockfile,
472            max_storage_size: Default::default(),
473            gclock: Default::default(),
474            mfs_root: Default::default(),
475        };
476        Repo {
477            inner: Arc::new(inner),
478        }
479    }
480
481    pub fn set_max_storage_size(&self, size: usize) {
482        self.inner.max_storage_size.store(size, Ordering::SeqCst);
483    }
484
485    pub fn max_storage_size(&self) -> usize {
486        self.inner.max_storage_size.load(Ordering::SeqCst)
487    }
488
489    pub async fn migrate(&self, repo: &Self) -> Result<(), Error> {
490        if self.is_online() || repo.is_online() {
491            anyhow::bail!("Repository cannot be online");
492        }
493        let block_migration = {
494            async move {
495                let mut errors = 0usize;
496                let mut stream = self.list_blocks().await;
497                while let Some(cid) = stream.next().await {
498                    match self.get_block_now(&cid).await {
499                        Ok(Some(block)) => {
500                            if let Err(e) = repo.inner.block_store.put(&block).await {
501                                error!("Error migrating {cid}: {e}");
502                                errors += 1;
503                            }
504                        }
505                        Ok(None) => {
506                            error!("{cid} doesnt exist");
507                            errors += 1;
508                        }
509                        Err(e) => {
510                            error!("Error getting block {cid}: {e}");
511                            errors += 1;
512                        }
513                    }
514                }
515                errors
516            }
517        };
518
519        let data_migration = {
520            async move {
521                let mut errors = 0usize;
522                let mut data_stream = self.data_store().iter().await;
523                while let Some((k, v)) = data_stream.next().await {
524                    if let Err(e) = repo.data_store().put(&k, &v).await {
525                        error!("Unable to migrate {k:?} into repo: {e}");
526                        errors += 1;
527                    }
528                }
529                errors
530            }
531        };
532
533        let pins_migration = {
534            async move {
535                let mut errors = 0usize;
536                let mut stream = self.data_store().list(None).await;
537                while let Some(item) = stream.next().await {
538                    let (cid, pin_mode) = match item {
539                        Ok(pin) => pin,
540                        Err(e) => {
541                            error!("Error listing pins during migration: {e}");
542                            errors += 1;
543                            continue;
544                        }
545                    };
546                    match pin_mode {
547                        PinMode::Direct => {
548                            if let Err(e) = repo.data_store().insert_direct_pin(&cid).await {
549                                error!("Unable to migrate pin {cid}: {e}");
550                                errors += 1;
551                            }
552                        }
553                        PinMode::Indirect => {
554                            //No need to track since we will be obtaining the reference from the pin that is recursive
555                            continue;
556                        }
557                        PinMode::Recursive => {
558                            let block = match self
559                                .get_block_now(&cid)
560                                .await
561                                .map(|block| block.and_then(|block| block.to_ipld().ok()))
562                            {
563                                Ok(Some(block)) => block,
564                                Ok(None) => continue,
565                                Err(e) => {
566                                    error!("Block {cid} does not exist but is pinned: {e}");
567                                    errors += 1;
568                                    continue;
569                                }
570                            };
571
572                            let st = crate::refs::IpldRefs::default()
573                                .with_only_unique()
574                                .refs_of_resolved(self, vec![(cid, block.clone())].into_iter())
575                                .map_ok(|crate::refs::Edge { destination, .. }| destination)
576                                .into_stream()
577                                .boxed();
578
579                            if let Err(e) = repo.insert_recursive_pin(&cid, st).await {
580                                error!("Error migrating pin {cid}: {e}");
581                                errors += 1;
582                                continue;
583                            }
584                        }
585                    }
586                }
587                errors
588            }
589        };
590
591        let (block_errors, data_errors, pin_errors) =
592            futures::join!(block_migration, data_migration, pins_migration);
593
594        let total = block_errors + data_errors + pin_errors;
595        if total > 0 {
596            anyhow::bail!(
597                "migration incomplete: {block_errors} block, {data_errors} datastore, {pin_errors} pin error(s); see logs"
598            );
599        }
600
601        Ok(())
602    }
603
604    pub(crate) fn initialize_channel(&self) -> Receiver<RepoEvent> {
605        let mut event_guard = self.inner.events.write();
606        let (sender, receiver) = channel(1);
607        debug_assert!(event_guard.is_none());
608        *event_guard = Some(sender);
609        self.set_online();
610        receiver
611    }
612
613    /// Shutdowns the repo, cancelling any pending subscriptions; Likely going away after some
614    /// refactoring, see notes on [`crate::Ipfs::exit_daemon`].
615    pub fn shutdown(&self) {
616        let mut map = self.inner.subscriptions.lock();
617        map.clear();
618        drop(map);
619        if let Some(mut event) = self.inner.events.write().take() {
620            event.close_channel()
621        }
622        self.set_offline();
623    }
624
625    pub fn is_online(&self) -> bool {
626        self.inner.online.load(Ordering::SeqCst)
627    }
628
629    pub(crate) fn set_online(&self) {
630        if self.is_online() {
631            return;
632        }
633
634        self.inner.online.store(true, Ordering::SeqCst)
635    }
636
637    pub(crate) fn set_offline(&self) {
638        if !self.is_online() {
639            return;
640        }
641
642        self.inner.online.store(false, Ordering::SeqCst)
643    }
644
645    fn repo_channel(&self) -> Option<Sender<RepoEvent>> {
646        self.inner.events.read().clone()
647    }
648
649    pub async fn init(&self) -> Result<(), Error> {
650        //Avoid initializing again
651        if self.inner.initialized.load(Ordering::SeqCst) {
652            return Ok(());
653        }
654        // Dropping the guard (even though not strictly necessary to compile) to avoid potential
655        // deadlocks if `block_store` or `data_store` were to try to access `Repo.lockfile`.
656        {
657            tracing::debug!("Trying lockfile");
658            self.inner.lockfile.try_exclusive()?;
659            tracing::debug!("lockfile tried");
660        }
661
662        self.inner.block_store.init().await?;
663        self.inner.data_store.init().await?;
664        self.inner.initialized.store(true, Ordering::SeqCst);
665        Ok(())
666    }
667
668    /// Puts a block into the block store.
669    pub fn put_block(&self, block: &Block) -> RepoPutBlock<S> {
670        RepoPutBlock::new(self, block).broadcast_on_new_block(true)
671    }
672
673    /// Puts multiple blocks into the block store in one batch, returning their cids in input order.
674    pub async fn put_blocks(&self, blocks: Vec<Block>) -> Result<Vec<Cid>, Error> {
675        if blocks.is_empty() {
676            return Ok(Vec::new());
677        }
678
679        let _guard = self.inner.gclock.read().await;
680        let results = self.inner.block_store.put_many(&blocks).await?;
681
682        let mut cids = Vec::with_capacity(results.len());
683        for ((cid, res), block) in results.into_iter().zip(blocks.iter()) {
684            if let BlockPut::NewBlock = res {
685                if let Some(mut event) = self.repo_channel() {
686                    _ = event.send(RepoEvent::NewBlock(block.clone())).await;
687                }
688                let list = self.inner.subscriptions.lock().remove(&cid);
689                if let Some(list) = list {
690                    for (_token, ch) in list {
691                        let _ = ch.send(Ok(block.clone()));
692                    }
693                }
694            }
695            cids.push(cid);
696        }
697
698        Ok(cids)
699    }
700
701    /// Retrives a block from the block store, or starts fetching it from the network and awaits
702    /// until it has been fetched.
703    #[inline]
704    pub fn get_block<C: Borrow<Cid>>(&self, cid: C) -> RepoGetBlock<S> {
705        RepoGetBlock::new(Repo::clone(self), cid)
706    }
707
708    /// Retrives a set of blocks from the block store, or starts fetching them from the network and awaits
709    /// until it has been fetched.
710    #[inline]
711    pub fn get_blocks(&self, cids: impl IntoIterator<Item = impl Borrow<Cid>>) -> RepoGetBlocks<S> {
712        RepoGetBlocks::new(Repo::clone(self)).blocks(cids)
713    }
714
715    /// Get the size of listed blocks
716    #[inline]
717    pub async fn get_blocks_size(&self, cids: &[Cid]) -> Result<Option<usize>, Error> {
718        self.inner.block_store.size(cids).await
719    }
720
721    /// Get the total size of the block store
722    #[inline]
723    pub async fn get_total_size(&self) -> Result<usize, Error> {
724        self.inner.block_store.total_size().await
725    }
726
727    /// Retrieves a block from the block store if it's available locally.
728    pub async fn get_block_now<C: Borrow<Cid>>(&self, cid: C) -> Result<Option<Block>, Error> {
729        let cid = cid.borrow();
730        self.inner.block_store.get(cid).await
731    }
732
733    /// Check to determine if blockstore contain a block
734    pub async fn contains<C: Borrow<Cid>>(&self, cid: C) -> Result<bool, Error> {
735        let cid = cid.borrow();
736        self.inner.block_store.contains(cid).await
737    }
738
739    /// Lists the blocks in the blockstore.
740    pub async fn list_blocks(&self) -> BoxStream<'static, Cid> {
741        self.inner.block_store.list().await
742    }
743
744    /// Remove block from the block store.
745    pub async fn remove_block<C: Borrow<Cid>>(
746        &self,
747        cid: C,
748        recursive: bool,
749    ) -> Result<Vec<Cid>, Error> {
750        let _guard = self.inner.gclock.read().await;
751        let cid = cid.borrow();
752        if self.is_pinned(cid).await? {
753            return Err(anyhow::anyhow!("block to remove is pinned"));
754        }
755
756        let list = match recursive {
757            true => {
758                let mut list = self.recursive_collections(*cid).await;
759                // ensure the first root block is apart of the list
760                list.insert(*cid);
761                list
762            }
763            false => BTreeSet::from_iter(std::iter::once(*cid)),
764        };
765
766        let list = stream::iter(
767            FuturesOrdered::from_iter(list.into_iter().map(|cid| async move { cid }))
768                .filter_map(|cid| async move {
769                    matches!(self.is_pinned(&cid).await, Ok(false)).then_some(cid)
770                })
771                .collect::<Vec<Cid>>()
772                .await,
773        )
774        .boxed();
775
776        let removed = self
777            .inner
778            .block_store
779            .remove_many(list)
780            .await
781            .collect::<Vec<_>>()
782            .await;
783
784        for cid in &removed {
785            // notify ipfs task about the removed blocks
786            if let Some(mut events) = self.repo_channel() {
787                let _ = events.send(RepoEvent::RemovedBlock(*cid)).await;
788            }
789        }
790        Ok(removed)
791    }
792
793    /// Collects the transitive references of `cid` (excluding `cid` itself). The accumulator doubles
794    /// as a visited set, so shared subtrees are walked once and any malformed cycle terminates.
795    fn recursive_collections(&self, cid: Cid) -> BoxFuture<'_, BTreeSet<Cid>> {
796        async move {
797            let mut visited = BTreeSet::new();
798            visited.insert(cid);
799            self.collect_references(cid, &mut visited).await;
800            visited.remove(&cid);
801            visited
802        }
803        .boxed()
804    }
805
806    fn collect_references<'a>(
807        &'a self,
808        cid: Cid,
809        visited: &'a mut BTreeSet<Cid>,
810    ) -> BoxFuture<'a, ()> {
811        async move {
812            let block = match self.get_block_now(&cid).await {
813                Ok(Some(block)) => block,
814                _ => return,
815            };
816
817            let mut references: BTreeSet<Cid> = BTreeSet::new();
818            if block.references(&mut references).is_err() {
819                return;
820            }
821
822            for reference in references {
823                if visited.insert(reference) {
824                    self.collect_references(reference, visited).await;
825                }
826            }
827        }
828        .boxed()
829    }
830
831    /// Pins a given Cid recursively or directly (non-recursively).
832    ///
833    /// Pins on a block are additive in sense that a previously directly (non-recursively) pinned
834    /// can be made recursive, but removing the recursive pin on the block removes also the direct
835    /// pin as well.
836    ///
837    /// Pinning a Cid recursively (for supported dag-protobuf and dag-cbor) will walk its
838    /// references and pin the references indirectly. When a Cid is pinned indirectly it will keep
839    /// its previous direct or recursive pin and be indirect in addition.
840    ///
841    /// Recursively pinned Cids cannot be re-pinned non-recursively but non-recursively pinned Cids
842    /// can be "upgraded to" being recursively pinned.
843    pub fn pin<C: Borrow<Cid>>(&self, cid: C) -> RepoInsertPin<S> {
844        RepoInsertPin::new(Repo::clone(self), cid)
845    }
846
847    /// Unpins a given Cid recursively or only directly.
848    ///
849    /// Recursively unpinning a previously only directly pinned Cid will remove the direct pin.
850    ///
851    /// Unpinning an indirectly pinned Cid is not possible other than through its recursively
852    /// pinned tree roots.
853    pub fn remove_pin<C: Borrow<Cid>>(&self, cid: C) -> RepoRemovePin<S> {
854        RepoRemovePin::new(Repo::clone(self), cid)
855    }
856
857    pub fn fetch<C: Borrow<Cid>>(&self, cid: C) -> RepoFetch<S> {
858        RepoFetch::new(Repo::clone(self), cid)
859    }
860
861    /// Pins a given Cid recursively or directly (non-recursively).
862    pub(crate) async fn insert_pin(
863        &self,
864        cid: &Cid,
865        recursive: bool,
866        local_only: bool,
867    ) -> Result<(), Error> {
868        let mut pin_fut = self.pin(cid);
869        if recursive {
870            pin_fut = pin_fut.recursive();
871        }
872        if local_only {
873            pin_fut = pin_fut.local();
874        }
875        pin_fut.await
876    }
877
878    /// Inserts a direct pin for a `Cid`.
879    pub(crate) async fn insert_direct_pin(&self, cid: &Cid) -> Result<(), Error> {
880        self.inner.data_store.insert_direct_pin(cid).await
881    }
882
883    /// Inserts a recursive pin for a `Cid`.
884    pub(crate) async fn insert_recursive_pin(
885        &self,
886        cid: &Cid,
887        refs: References<'_>,
888    ) -> Result<(), Error> {
889        self.inner.data_store.insert_recursive_pin(cid, refs).await
890    }
891
892    /// Removes a direct pin for a `Cid`.
893    pub(crate) async fn remove_direct_pin(&self, cid: &Cid) -> Result<(), Error> {
894        self.inner.data_store.remove_direct_pin(cid).await
895    }
896
897    /// Removes a recursive pin for a `Cid`.
898    pub(crate) async fn remove_recursive_pin(
899        &self,
900        cid: &Cid,
901        refs: References<'_>,
902    ) -> Result<(), Error> {
903        // FIXME: not really sure why is there not an easier way to to transfer control
904        self.inner.data_store.remove_recursive_pin(cid, refs).await
905    }
906
907    /// Function to perform a basic cleanup of unpinned blocks
908    pub(crate) async fn cleanup(&self) -> Result<Vec<Cid>, Error> {
909        let repo = self.clone();
910
911        let blocks = repo.list_blocks().await;
912        let pins = repo
913            .list_pins(None)
914            .await
915            .map_ok(|(cid, _)| cid)
916            .try_collect::<HashSet<Cid>>()
917            .await?;
918
919        let stream = async_stream::stream! {
920            for await cid in blocks {
921                if pins.contains(&cid) {
922                    continue;
923                }
924                yield cid;
925            }
926        }
927        .boxed();
928
929        let removed_blocks = self
930            .inner
931            .block_store
932            .remove_many(stream)
933            .await
934            .collect::<Vec<_>>()
935            .await;
936
937        Ok(removed_blocks)
938    }
939
940    /// Checks if a `Cid` is pinned.
941    pub async fn is_pinned<C: Borrow<Cid>>(&self, cid: C) -> Result<bool, Error> {
942        let cid = cid.borrow();
943        self.inner.data_store.is_pinned(cid).await
944    }
945
946    pub async fn list_pins(
947        &self,
948        mode: impl Into<Option<PinMode>>,
949    ) -> BoxStream<'static, Result<(Cid, PinMode), Error>> {
950        let mode = mode.into();
951        self.inner.data_store.list(mode).await
952    }
953
954    pub async fn query_pins(
955        &self,
956        cids: Vec<Cid>,
957        requirement: impl Into<Option<PinMode>>,
958    ) -> Result<Vec<(Cid, PinKind<Cid>)>, Error> {
959        let requirement = requirement.into();
960        self.inner.data_store.query(cids, requirement).await
961    }
962}
963
964pub struct GCGuard<'a> {
965    _g: RwLockReadGuard<'a, ()>,
966}
967
968impl<S: RepoTypes> Repo<S> {
969    /// Hold a guard to prevent GC from running until this guard has dropped
970    /// Note: Until this guard drops, the GC task, if enabled, would not perform any cleanup.
971    ///       If the GC task is running, this guard will await until GC finishes
972    pub async fn gc_guard(&self) -> GCGuard<'_> {
973        let _g = self.inner.gclock.read().await;
974        GCGuard { _g }
975    }
976
977    pub fn data_store(&self) -> &S::TDataStore {
978        &self.inner.data_store
979    }
980}
981
982pub struct RepoGetBlock<S: RepoTypes> {
983    instance: RepoGetBlocks<S>,
984}
985
986impl<S: RepoTypes> RepoGetBlock<S> {
987    pub fn new(repo: Repo<S>, cid: impl Borrow<Cid>) -> Self {
988        let instance = RepoGetBlocks::new(repo).block(cid);
989        Self { instance }
990    }
991
992    pub fn span(mut self, span: impl Borrow<Span>) -> Self {
993        self.instance = self.instance.span(span);
994        self
995    }
996
997    pub fn timeout(mut self, timeout: impl Into<Option<Duration>>) -> Self {
998        self.instance = self.instance.timeout(timeout);
999        self
1000    }
1001
1002    pub fn local(mut self) -> Self {
1003        self.instance = self.instance.local();
1004        self
1005    }
1006
1007    pub fn set_local(mut self, local: bool) -> Self {
1008        self.instance = self.instance.set_local(local);
1009        self
1010    }
1011
1012    /// Peer that may contain the block
1013    pub fn provider(mut self, peer_id: PeerId) -> Self {
1014        self.instance = self.instance.provider(peer_id);
1015        self
1016    }
1017
1018    /// List of peers that may contain the block
1019    pub fn providers(mut self, providers: impl IntoIterator<Item = impl Borrow<PeerId>>) -> Self {
1020        self.instance = self.instance.providers(providers);
1021        self
1022    }
1023}
1024
1025impl<S: RepoTypes> Future for RepoGetBlock<S> {
1026    type Output = Result<Block, Error>;
1027    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
1028        let this = &mut self;
1029        match futures::ready!(this.instance.poll_next_unpin(cx)) {
1030            Some(result) => Poll::Ready(result),
1031            None => Poll::Ready(Err(anyhow::anyhow!("block does not exist"))),
1032        }
1033    }
1034}
1035
1036pub struct RepoGetBlocks<S: RepoTypes> {
1037    repo: Option<Repo<S>>,
1038    cids: IndexSet<Cid>,
1039    providers: IndexSet<PeerId>,
1040    local: bool,
1041    span: Span,
1042    timeout: Option<Duration>,
1043    stream: Option<BoxStream<'static, Result<Block, Error>>>,
1044}
1045
1046impl<S: RepoTypes> RepoGetBlocks<S> {
1047    pub fn new(repo: Repo<S>) -> Self {
1048        Self {
1049            repo: Some(repo),
1050            cids: IndexSet::new(),
1051            providers: IndexSet::new(),
1052            local: false,
1053            span: Span::current(),
1054            timeout: None,
1055            stream: None,
1056        }
1057    }
1058
1059    pub fn blocks(mut self, cids: impl IntoIterator<Item = impl Borrow<Cid>>) -> Self {
1060        self.cids.extend(cids.into_iter().map(|cid| *cid.borrow()));
1061        self
1062    }
1063
1064    pub fn block(self, cid: impl Borrow<Cid>) -> Self {
1065        self.blocks([cid])
1066    }
1067
1068    pub fn span(mut self, span: impl Borrow<Span>) -> Self {
1069        let span = span.borrow();
1070        self.span = span.clone();
1071        self
1072    }
1073
1074    pub fn timeout(mut self, timeout: impl Into<Option<Duration>>) -> Self {
1075        self.timeout = timeout.into();
1076        self
1077    }
1078
1079    pub fn local(mut self) -> Self {
1080        self.local = true;
1081        self
1082    }
1083
1084    pub fn set_local(mut self, local: bool) -> Self {
1085        self.local = local;
1086        self
1087    }
1088
1089    pub fn provider(self, peer_id: impl Borrow<PeerId>) -> Self {
1090        self.providers([peer_id])
1091    }
1092
1093    pub fn providers(mut self, providers: impl IntoIterator<Item = impl Borrow<PeerId>>) -> Self {
1094        self.providers
1095            .extend(providers.into_iter().map(|k| *k.borrow()));
1096        self
1097    }
1098}
1099
1100impl<S: RepoTypes> Stream for RepoGetBlocks<S> {
1101    type Item = Result<Block, Error>;
1102
1103    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
1104        if self.stream.is_none() && self.repo.is_none() {
1105            return Poll::Ready(None);
1106        }
1107
1108        let this = &mut *self;
1109
1110        loop {
1111            match &mut this.stream {
1112                Some(stream) => {
1113                    let _g = this.span.enter();
1114                    match futures::ready!(stream.poll_next_unpin(cx)) {
1115                        Some(item) => return Poll::Ready(Some(item)),
1116                        None => {
1117                            this.stream.take();
1118                            return Poll::Ready(None);
1119                        }
1120                    }
1121                }
1122                None => {
1123                    let repo = this.repo.take().expect("valid repo instance");
1124                    let providers = std::mem::take(&mut this.providers);
1125                    let cids = std::mem::take(&mut this.cids);
1126                    let local_only = this.local;
1127                    let timeout = this.timeout;
1128
1129                    let st = async_stream::stream! {
1130                        let _guard = repo.gc_guard().await;
1131                        let mut missing: IndexSet<Cid> = cids.clone();
1132                        for cid in &cids {
1133                            if let Ok(Some(block)) = repo.get_block_now(cid).await {
1134                                yield Ok(block);
1135                                missing.shift_remove(cid);
1136                            }
1137                        }
1138
1139                        if missing.is_empty() {
1140                            return;
1141                        }
1142
1143                        if local_only || !repo.is_online() {
1144                            yield Err(anyhow::anyhow!("Unable to locate missing blocks {missing:?}"));
1145                            return;
1146                        }
1147
1148                        let mut events = match repo.repo_channel() {
1149                            Some(events) => events,
1150                            None => {
1151                                yield Err(anyhow::anyhow!("Channel is not available"));
1152                                return;
1153                            }
1154                        };
1155
1156                        let timeout = timeout.or(Some(Duration::from_secs(60)));
1157
1158                        let blocks = FuturesUnordered::new();
1159                        let mut wants = Vec::with_capacity(missing.len());
1160
1161                        for cid in &missing {
1162                            let cid = *cid;
1163                            let token = SUBSCRIPTION_TOKEN.fetch_add(1, Ordering::Relaxed);
1164                            let (tx, rx) = futures::channel::oneshot::channel();
1165                            repo.inner
1166                                .subscriptions
1167                                .lock()
1168                                .entry(cid)
1169                                .or_default()
1170                                .insert(token, tx);
1171
1172                            let guard = WaiterGuard {
1173                                repo: repo.clone(),
1174                                events: events.clone(),
1175                                cid,
1176                                token,
1177                            };
1178
1179                            let task = async move {
1180                                // unregisters this waiter and emits UnwantBlock on every exit,
1181                                // including this future being dropped before completion.
1182                                let _guard = guard;
1183
1184                                // deadline so an unavailable block errors instead of hanging forever
1185                                let timeout_fut = async move {
1186                                    match timeout {
1187                                        Some(duration) => futures_timer::Delay::new(duration).await,
1188                                        None => futures::future::pending::<()>().await,
1189                                    }
1190                                };
1191                                futures::pin_mut!(timeout_fut);
1192
1193                                match futures::future::select(rx, timeout_fut).await {
1194                                    Either::Left((Ok(Ok(block)), _)) => Ok::<_, Error>(block),
1195                                    Either::Left((Ok(Err(e)), _)) => Err::<_, Error>(anyhow::anyhow!("{e}")),
1196                                    Either::Left((Err(e), _)) => Err::<_, Error>(e.into()),
1197                                    Either::Right(((), _)) => {
1198                                        Err::<_, Error>(anyhow::anyhow!("request for {cid} timed out"))
1199                                    }
1200                                }
1201                            }
1202                            .boxed();
1203
1204                            wants.push(cid);
1205                            blocks.push(task);
1206                        }
1207
1208                        events
1209                            .send(RepoEvent::WantBlock(
1210                                wants,
1211                                Vec::from_iter(providers),
1212                                timeout,
1213                            ))
1214                            .await
1215                            .ok();
1216
1217                        for await block in blocks {
1218                            yield block
1219                        }
1220                    };
1221
1222                    this.stream.replace(Box::pin(st));
1223                }
1224            }
1225        }
1226    }
1227}
1228
1229impl<S: RepoTypes> IntoFuture for RepoGetBlocks<S> {
1230    type Output = Result<Vec<Block>, Error>;
1231    type IntoFuture = BoxFuture<'static, Self::Output>;
1232    fn into_future(self) -> Self::IntoFuture {
1233        async move {
1234            let col = self.try_collect().await?;
1235            Ok(col)
1236        }
1237        .boxed()
1238    }
1239}
1240
1241pub struct RepoPutBlock<S: RepoTypes> {
1242    repo: Repo<S>,
1243    block: Option<Block>,
1244    span: Option<Span>,
1245    broadcast_on_new_block: bool,
1246}
1247
1248impl<S: RepoTypes> RepoPutBlock<S> {
1249    fn new(repo: &Repo<S>, block: &Block) -> Self {
1250        let block = Some(block.clone());
1251        Self {
1252            repo: Repo::clone(repo),
1253            block,
1254            span: None,
1255            broadcast_on_new_block: true,
1256        }
1257    }
1258
1259    pub fn broadcast_on_new_block(mut self, v: bool) -> Self {
1260        self.broadcast_on_new_block = v;
1261        self
1262    }
1263
1264    pub fn span(mut self, span: Span) -> Self {
1265        self.span = Some(span);
1266        self
1267    }
1268}
1269
1270impl<S: RepoTypes> IntoFuture for RepoPutBlock<S> {
1271    type IntoFuture = BoxFuture<'static, Self::Output>;
1272    type Output = Result<Cid, Error>;
1273    fn into_future(mut self) -> Self::IntoFuture {
1274        let block = self.block.take().expect("valid block is set");
1275        let span = self.span.unwrap_or(Span::current());
1276        let span = debug_span!(parent: &span, "put_block", cid = %block.cid());
1277        async move {
1278            let _guard = self.repo.inner.gclock.read().await;
1279            let (cid, res) = self.repo.inner.block_store.put(&block).await?;
1280
1281            if let BlockPut::NewBlock = res {
1282                if self.broadcast_on_new_block {
1283                    if let Some(mut event) = self.repo.repo_channel() {
1284                        _ = event.send(RepoEvent::NewBlock(block.clone())).await;
1285                    }
1286                }
1287                let list = self.repo.inner.subscriptions.lock().remove(&cid);
1288                if let Some(list) = list {
1289                    for (_token, ch) in list {
1290                        let _ = ch.send(Ok(block.clone()));
1291                    }
1292                }
1293            }
1294
1295            Ok(cid)
1296        }
1297        .instrument(span)
1298        .boxed()
1299    }
1300}
1301
1302pub struct RepoFetch<S: RepoTypes> {
1303    repo: Repo<S>,
1304    cid: Cid,
1305    span: Option<Span>,
1306    providers: Vec<PeerId>,
1307    recursive: bool,
1308    timeout: Option<Duration>,
1309    refs: crate::refs::IpldRefs,
1310}
1311
1312impl<S: RepoTypes> RepoFetch<S> {
1313    pub fn new<C: Borrow<Cid>>(repo: Repo<S>, cid: C) -> Self {
1314        let cid = cid.borrow();
1315        Self {
1316            repo,
1317            cid: *cid,
1318            recursive: false,
1319            providers: vec![],
1320            timeout: None,
1321            refs: Default::default(),
1322            span: None,
1323        }
1324    }
1325
1326    /// Fetch blocks recursively
1327    pub fn recursive(mut self) -> Self {
1328        self.recursive = true;
1329        self
1330    }
1331
1332    /// Peer that may contain the block
1333    pub fn provider(mut self, peer_id: PeerId) -> Self {
1334        if !self.providers.contains(&peer_id) {
1335            self.providers.push(peer_id);
1336        }
1337        self
1338    }
1339
1340    /// List of peers that may contain the block
1341    pub fn providers(mut self, providers: &[PeerId]) -> Self {
1342        self.providers = providers.to_vec();
1343        self
1344    }
1345
1346    /// Fetch blocks to a specific depth
1347    pub fn depth(mut self, depth: u64) -> Self {
1348        self.refs = self.refs.with_max_depth(depth);
1349        self
1350    }
1351
1352    /// Duration to fetch the block from the network before
1353    /// timing out
1354    pub fn timeout(mut self, duration: Duration) -> Self {
1355        self.timeout.replace(duration);
1356        self.refs = self.refs.with_timeout(duration);
1357        self
1358    }
1359
1360    pub fn exit_on_error(mut self) -> Self {
1361        self.refs = self.refs.with_exit_on_error();
1362        self
1363    }
1364
1365    /// Set tracing span
1366    pub fn span(mut self, span: Span) -> Self {
1367        self.span = Some(span);
1368        self
1369    }
1370}
1371
1372impl<S: RepoTypes> IntoFuture for RepoFetch<S> {
1373    type Output = Result<(), Error>;
1374
1375    type IntoFuture = BoxFuture<'static, Self::Output>;
1376
1377    fn into_future(self) -> Self::IntoFuture {
1378        let cid = self.cid;
1379        let span = self.span.unwrap_or(Span::current());
1380        let recursive = self.recursive;
1381        let repo = self.repo;
1382        let span = debug_span!(parent: &span, "fetch", cid = %cid, recursive);
1383        let providers = self.providers;
1384        let timeout = self.timeout;
1385        async move {
1386            // Although getting a block adds a guard, we will add a read guard here a head of time so we can hold it throughout this future
1387            let _g = repo.inner.gclock.read().await;
1388            let block = repo
1389                .get_block(cid)
1390                .providers(&providers)
1391                .timeout(timeout)
1392                .await?;
1393
1394            if !recursive {
1395                return Ok(());
1396            }
1397            let ipld = block.to_ipld()?;
1398
1399            let mut st = self
1400                .refs
1401                .with_only_unique()
1402                .providers(&providers)
1403                .refs_of_resolved(&repo, vec![(cid, ipld.clone())])
1404                .map_ok(|crate::refs::Edge { destination, .. }| destination)
1405                .into_stream()
1406                .boxed();
1407
1408            while let Some(_c) = st.try_next().await? {}
1409
1410            Ok(())
1411        }
1412        .instrument(span)
1413        .boxed()
1414    }
1415}
1416
1417pub struct RepoInsertPin<S: RepoTypes> {
1418    repo: Repo<S>,
1419    cid: Cid,
1420    span: Option<Span>,
1421    providers: Vec<PeerId>,
1422    recursive: bool,
1423    timeout: Option<Duration>,
1424    local: bool,
1425    refs: crate::refs::IpldRefs,
1426}
1427
1428impl<S: RepoTypes> RepoInsertPin<S> {
1429    pub fn new<C: Borrow<Cid>>(repo: Repo<S>, cid: C) -> Self {
1430        let cid = cid.borrow();
1431        Self {
1432            repo,
1433            cid: *cid,
1434            recursive: false,
1435            providers: vec![],
1436            local: false,
1437            timeout: None,
1438            refs: Default::default(),
1439            span: None,
1440        }
1441    }
1442
1443    /// Recursively pin blocks
1444    pub fn recursive(mut self) -> Self {
1445        self.recursive = true;
1446        self
1447    }
1448
1449    /// Pin local blocks only
1450    pub fn local(mut self) -> Self {
1451        self.local = true;
1452        self.refs = self.refs.with_existing_blocks();
1453        self
1454    }
1455
1456    /// Set a flag to pin local blocks only
1457    pub fn set_local(mut self, local: bool) -> Self {
1458        self.local = local;
1459        if local {
1460            self.refs = self.refs.with_existing_blocks();
1461        }
1462        self
1463    }
1464
1465    /// Peer that may contain the block to pin
1466    pub fn provider(mut self, peer_id: PeerId) -> Self {
1467        if !self.providers.contains(&peer_id) {
1468            self.providers.push(peer_id);
1469        }
1470        self
1471    }
1472
1473    /// List of peers that may contain the block to pin
1474    pub fn providers(mut self, providers: &[PeerId]) -> Self {
1475        self.providers = providers.into();
1476        self
1477    }
1478
1479    /// Pin to a specific depth of the graph
1480    pub fn depth(mut self, depth: u64) -> Self {
1481        self.refs = self.refs.with_max_depth(depth);
1482        self
1483    }
1484
1485    /// Duration to fetch the block from the network before
1486    /// timing out
1487    pub fn timeout(mut self, duration: Duration) -> Self {
1488        self.timeout.replace(duration);
1489        self.refs = self.refs.with_timeout(duration);
1490        self
1491    }
1492
1493    pub fn exit_on_error(mut self) -> Self {
1494        self.refs = self.refs.with_exit_on_error();
1495        self
1496    }
1497
1498    /// Set tracing span
1499    pub fn span(mut self, span: Span) -> Self {
1500        self.span = Some(span);
1501        self
1502    }
1503}
1504
1505impl<S: RepoTypes> IntoFuture for RepoInsertPin<S> {
1506    type Output = Result<(), Error>;
1507
1508    type IntoFuture = BoxFuture<'static, Self::Output>;
1509
1510    fn into_future(self) -> Self::IntoFuture {
1511        let cid = self.cid;
1512        let local = self.local;
1513        let span = self.span.unwrap_or(Span::current());
1514        let recursive = self.recursive;
1515        let repo = self.repo;
1516        let span = debug_span!(parent: &span, "insert_pin", cid = %cid, recursive);
1517        let providers = self.providers;
1518        let timeout = self.timeout;
1519        async move {
1520            // Although getting a block adds a guard, we will add a read guard here a head of time so we can hold it throughout this future
1521            let _g = repo.inner.gclock.read().await;
1522            let block = repo
1523                .get_block(cid)
1524                .providers(&providers)
1525                .set_local(local)
1526                .timeout(timeout)
1527                .await?;
1528
1529            if !recursive {
1530                repo.insert_direct_pin(&cid).await?
1531            } else {
1532                let ipld = block.to_ipld()?;
1533
1534                let st = self
1535                    .refs
1536                    .with_only_unique()
1537                    .providers(&providers)
1538                    .refs_of_resolved(&repo, vec![(cid, ipld.clone())])
1539                    .map_ok(|crate::refs::Edge { destination, .. }| destination)
1540                    .into_stream()
1541                    .boxed();
1542
1543                repo.insert_recursive_pin(&cid, st).await?
1544            }
1545            Ok(())
1546        }
1547        .instrument(span)
1548        .boxed()
1549    }
1550}
1551
1552pub struct RepoRemovePin<S: RepoTypes> {
1553    repo: Repo<S>,
1554    cid: Cid,
1555    span: Option<Span>,
1556    recursive: bool,
1557    refs: crate::refs::IpldRefs,
1558}
1559
1560impl<S: RepoTypes> RepoRemovePin<S> {
1561    pub fn new<C: Borrow<Cid>>(repo: Repo<S>, cid: C) -> Self {
1562        let cid = cid.borrow();
1563        Self {
1564            repo,
1565            cid: *cid,
1566            recursive: false,
1567            refs: Default::default(),
1568            span: None,
1569        }
1570    }
1571
1572    /// Recursively unpin blocks
1573    pub fn recursive(mut self) -> Self {
1574        self.recursive = true;
1575        self
1576    }
1577
1578    /// Set tracing span
1579    pub fn span(mut self, span: Span) -> Self {
1580        self.span = Some(span);
1581        self
1582    }
1583}
1584
1585impl<S: RepoTypes> IntoFuture for RepoRemovePin<S> {
1586    type Output = Result<(), Error>;
1587
1588    type IntoFuture = BoxFuture<'static, Self::Output>;
1589
1590    fn into_future(self) -> Self::IntoFuture {
1591        let cid = self.cid;
1592        let span = self.span.unwrap_or(Span::current());
1593        let recursive = self.recursive;
1594        let repo = self.repo;
1595
1596        let span = debug_span!(parent: &span, "remove_pin", cid = %cid, recursive);
1597        async move {
1598            let _g = repo.inner.gclock.read().await;
1599            if !recursive {
1600                repo.remove_direct_pin(&cid).await
1601            } else {
1602                // start walking refs of the root after loading it
1603
1604                let block = match repo.get_block_now(&cid).await? {
1605                    Some(b) => b,
1606                    None => {
1607                        return Err(anyhow::anyhow!("pinned root not found: {}", cid));
1608                    }
1609                };
1610
1611                let ipld = block.to_ipld()?;
1612                let st = self
1613                    .refs
1614                    .with_only_unique()
1615                    .with_existing_blocks()
1616                    .refs_of_resolved(&repo, vec![(cid, ipld)])
1617                    .map_ok(|crate::refs::Edge { destination, .. }| destination)
1618                    .into_stream()
1619                    .boxed();
1620
1621                repo.remove_recursive_pin(&cid, st).await
1622            }
1623        }
1624        .instrument(span)
1625        .boxed()
1626    }
1627}
1628
1629#[cfg(test)]
1630mod repo_tests {
1631    use super::*;
1632    use ipld_core::codec::Codec;
1633    use ipld_core::ipld::Ipld;
1634    use multihash_codetable::{Code, MultihashDigest};
1635
1636    fn raw_block(data: &[u8]) -> Block {
1637        let cid = Cid::new_v1(0x55, Code::Sha2_256.digest(data));
1638        Block::new(cid, data.to_vec()).unwrap()
1639    }
1640
1641    fn dag_cbor(ipld: &Ipld) -> Block {
1642        let data = serde_ipld_dagcbor::codec::DagCborCodec::encode_to_vec(ipld).unwrap();
1643        let cid = Cid::new_v1(0x71, Code::Sha2_256.digest(&data));
1644        Block::new(cid, data).unwrap()
1645    }
1646
1647    #[tokio::test]
1648    async fn recursive_remove_walks_shared_subtree_once() {
1649        let repo = Repo::new_memory();
1650        repo.init().await.unwrap();
1651
1652        // diamond: root -> {a, b}, a -> shared, b -> shared
1653        let shared = dag_cbor(&Ipld::String("shared".into()));
1654        let a = dag_cbor(&Ipld::List(vec![
1655            Ipld::Integer(0),
1656            Ipld::Link(*shared.cid()),
1657        ]));
1658        let b = dag_cbor(&Ipld::List(vec![
1659            Ipld::Integer(1),
1660            Ipld::Link(*shared.cid()),
1661        ]));
1662        let root = dag_cbor(&Ipld::List(vec![
1663            Ipld::Link(*a.cid()),
1664            Ipld::Link(*b.cid()),
1665        ]));
1666
1667        for blk in [&shared, &a, &b, &root] {
1668            repo.put_block(blk).await.unwrap();
1669        }
1670
1671        let removed = repo.remove_block(*root.cid(), true).await.unwrap();
1672
1673        assert_eq!(
1674            removed.len(),
1675            4,
1676            "expected root+a+b+shared once: {removed:?}"
1677        );
1678        for blk in [&shared, &a, &b, &root] {
1679            assert!(
1680                !repo.contains(blk.cid()).await.unwrap(),
1681                "block still present: {}",
1682                blk.cid()
1683            );
1684        }
1685    }
1686
1687    #[tokio::test]
1688    async fn migrate_copies_blocks_and_pins() {
1689        let src = Repo::new_memory();
1690        src.init().await.unwrap();
1691        let dst = Repo::new_memory();
1692        dst.init().await.unwrap();
1693
1694        let direct = raw_block(b"direct pinned block");
1695        let leaf = dag_cbor(&Ipld::String("leaf".into()));
1696        let root = dag_cbor(&Ipld::List(vec![Ipld::Link(*leaf.cid())]));
1697
1698        for blk in [&direct, &leaf, &root] {
1699            src.put_block(blk).await.unwrap();
1700        }
1701        src.pin(*direct.cid()).await.unwrap();
1702        src.pin(*root.cid()).recursive().await.unwrap();
1703
1704        src.migrate(&dst).await.unwrap();
1705
1706        for blk in [&direct, &leaf, &root] {
1707            assert!(
1708                dst.contains(blk.cid()).await.unwrap(),
1709                "missing block {}",
1710                blk.cid()
1711            );
1712        }
1713        assert!(dst.is_pinned(direct.cid()).await.unwrap());
1714        assert!(dst.is_pinned(root.cid()).await.unwrap());
1715        assert!(
1716            dst.is_pinned(leaf.cid()).await.unwrap(),
1717            "leaf must be indirectly pinned after migrating the recursive pin"
1718        );
1719    }
1720}
rust_ipfs/repo/mod.rs

rust_ipfs/repo/
mod.rs
