iroh_blobs/store/
gc.rs

1use std::{collections::HashSet, pin::Pin, sync::Arc};
2
3use bao_tree::ChunkRanges;
4use genawaiter::sync::{Co, Gen};
5use n0_future::{time::Duration, Stream, StreamExt};
6use tracing::{debug, error, info, warn};
7
8use crate::{api::Store, Hash, HashAndFormat};
9
10/// An event related to GC
11#[derive(Debug)]
12pub enum GcMarkEvent {
13    /// A custom event (info)
14    CustomDebug(String),
15    /// A custom non critical error
16    CustomWarning(String, Option<crate::api::Error>),
17    /// An unrecoverable error during GC
18    Error(crate::api::Error),
19}
20
21/// An event related to GC
22#[derive(Debug)]
23pub enum GcSweepEvent {
24    /// A custom event (debug)
25    CustomDebug(String),
26    /// A custom non critical error
27    #[allow(dead_code)]
28    CustomWarning(String, Option<crate::api::Error>),
29    /// An unrecoverable error during GC
30    Error(crate::api::Error),
31}
32
33/// Compute the set of live hashes
34pub(super) async fn gc_mark_task(
35    store: &Store,
36    live: &mut HashSet<Hash>,
37    co: &Co<GcMarkEvent>,
38) -> crate::api::Result<()> {
39    macro_rules! trace {
40        ($($arg:tt)*) => {
41            co.yield_(GcMarkEvent::CustomDebug(format!($($arg)*))).await;
42        };
43    }
44    macro_rules! warn {
45        ($($arg:tt)*) => {
46            co.yield_(GcMarkEvent::CustomWarning(format!($($arg)*), None)).await;
47        };
48    }
49    let mut roots = HashSet::new();
50    trace!("traversing tags");
51    let mut tags = store.tags().list().await?;
52    while let Some(tag) = tags.next().await {
53        let info = tag?;
54        trace!("adding root {:?} {:?}", info.name, info.hash_and_format());
55        roots.insert(info.hash_and_format());
56    }
57    trace!("traversing temp roots");
58    let mut tts = store.tags().list_temp_tags().await?;
59    while let Some(tt) = tts.next().await {
60        trace!("adding temp root {:?}", tt);
61        roots.insert(tt);
62    }
63    for HashAndFormat { hash, format } in roots {
64        // we need to do this for all formats except raw
65        if live.insert(hash) && !format.is_raw() {
66            let mut stream = store.export_bao(hash, ChunkRanges::all()).hashes();
67            while let Some(hash) = stream.next().await {
68                match hash {
69                    Ok(hash) => {
70                        live.insert(hash);
71                    }
72                    Err(e) => {
73                        warn!("error while traversing hashseq: {e:?}");
74                    }
75                }
76            }
77        }
78    }
79    trace!("gc mark done. found {} live blobs", live.len());
80    Ok(())
81}
82
83async fn gc_sweep_task(
84    store: &Store,
85    live: &HashSet<Hash>,
86    co: &Co<GcSweepEvent>,
87) -> crate::api::Result<()> {
88    let mut blobs = store.blobs().list().stream().await?;
89    let mut count = 0;
90    let mut batch = Vec::new();
91    while let Some(hash) = blobs.next().await {
92        let hash = hash?;
93        if !live.contains(&hash) {
94            batch.push(hash);
95            count += 1;
96        }
97        if batch.len() >= 100 {
98            store.blobs().delete(batch.clone()).await?;
99            batch.clear();
100        }
101    }
102    if !batch.is_empty() {
103        store.blobs().delete(batch).await?;
104    }
105    store.sync_db().await?;
106    co.yield_(GcSweepEvent::CustomDebug(format!("deleted {count} blobs")))
107        .await;
108    Ok(())
109}
110
111fn gc_mark<'a>(
112    store: &'a Store,
113    live: &'a mut HashSet<Hash>,
114) -> impl Stream<Item = GcMarkEvent> + 'a {
115    Gen::new(|co| async move {
116        if let Err(e) = gc_mark_task(store, live, &co).await {
117            co.yield_(GcMarkEvent::Error(e)).await;
118        }
119    })
120}
121
122fn gc_sweep<'a>(
123    store: &'a Store,
124    live: &'a HashSet<Hash>,
125) -> impl Stream<Item = GcSweepEvent> + 'a {
126    Gen::new(|co| async move {
127        if let Err(e) = gc_sweep_task(store, live, &co).await {
128            co.yield_(GcSweepEvent::Error(e)).await;
129        }
130    })
131}
132
133/// Configuration for garbage collection.
134///
135/// To protect blobs during long-running writes without pausing the GC
136/// schedule, use [`crate::api::blobs::Batch::temp_tag`].
137#[derive(derive_more::Debug, Clone)]
138pub struct GcConfig {
139    /// Interval in which to run garbage collection.
140    pub interval: Duration,
141    /// Optional callback to manually add protected blobs.
142    ///
143    /// The callback is called before each garbage collection run. It gets a `&mut HashSet<Hash>`
144    /// and returns a future that returns [`ProtectOutcome`]. All hashes that are added to the
145    /// [`HashSet`] will be protected from garbage collection during this run.
146    ///
147    /// In normal operation, return [`ProtectOutcome::Continue`] from the callback. If you return
148    /// [`ProtectOutcome::Abort`], the garbage collection run will be aborted.Use this if your
149    /// source of hashes to protect returned an error, and thus garbage collection should be skipped
150    /// completely to not unintentionally delete blobs that should be protected.
151    #[debug("ProtectCallback")]
152    pub add_protected: Option<ProtectCb>,
153}
154
155/// Returned from [`ProtectCb`].
156///
157/// See [`GcConfig::add_protected] for details.
158#[derive(Debug)]
159pub enum ProtectOutcome {
160    /// Continue with the garbage collection run.
161    Continue,
162    /// Abort the garbage collection run.
163    Abort,
164}
165
166/// The type of the garbage collection callback.
167///
168/// See [`GcConfig::add_protected] for details.
169pub type ProtectCb = Arc<
170    dyn for<'a> Fn(
171            &'a mut HashSet<Hash>,
172        )
173            -> Pin<Box<dyn std::future::Future<Output = ProtectOutcome> + Send + Sync + 'a>>
174        + Send
175        + Sync
176        + 'static,
177>;
178
179pub async fn gc_run_once(store: &Store, live: &mut HashSet<Hash>) -> crate::api::Result<()> {
180    debug!(externally_protected = live.len(), "gc: start");
181    {
182        store.clear_protected().await?;
183        let mut stream = gc_mark(store, live);
184        while let Some(ev) = stream.next().await {
185            match ev {
186                GcMarkEvent::CustomDebug(msg) => {
187                    debug!("{}", msg);
188                }
189                GcMarkEvent::CustomWarning(msg, err) => {
190                    warn!("{}: {:?}", msg, err);
191                }
192                GcMarkEvent::Error(err) => {
193                    error!("error during gc mark: {:?}", err);
194                    return Err(err);
195                }
196            }
197        }
198    }
199    debug!(total_protected = live.len(), "gc: sweep");
200    {
201        let mut stream = gc_sweep(store, live);
202        while let Some(ev) = stream.next().await {
203            match ev {
204                GcSweepEvent::CustomDebug(msg) => {
205                    debug!("{}", msg);
206                }
207                GcSweepEvent::CustomWarning(msg, err) => {
208                    warn!("{}: {:?}", msg, err);
209                }
210                GcSweepEvent::Error(err) => {
211                    error!("error during gc sweep: {:?}", err);
212                    return Err(err);
213                }
214            }
215        }
216    }
217    debug!("gc: done");
218
219    Ok(())
220}
221
222pub async fn run_gc(store: Store, config: GcConfig) {
223    debug!("gc enabled with interval {:?}", config.interval);
224    let mut live = HashSet::new();
225    loop {
226        live.clear();
227        n0_future::time::sleep(config.interval).await;
228        if let Some(ref cb) = config.add_protected {
229            match (cb)(&mut live).await {
230                ProtectOutcome::Continue => {}
231                ProtectOutcome::Abort => {
232                    info!("abort gc run: protect callback indicated abort");
233                    continue;
234                }
235            }
236        }
237        if let Err(e) = gc_run_once(&store, &mut live).await {
238            error!("error during gc run: {e}");
239            break;
240        }
241    }
242}
243
244#[cfg(test)]
245mod tests {
246    use std::io::{self};
247
248    use bao_tree::io::EncodeError;
249    use range_collections::RangeSet2;
250    use testresult::TestResult;
251
252    use super::*;
253    use crate::{
254        api::{blobs::AddBytesOptions, ExportBaoError, RequestError, Store},
255        hashseq::HashSeq,
256        BlobFormat,
257    };
258
259    async fn gc_smoke(store: &Store) -> TestResult<()> {
260        let blobs = store.blobs();
261        let at = blobs.add_slice("a").temp_tag().await?;
262        let bt = blobs.add_slice("b").temp_tag().await?;
263        let ct = blobs.add_slice("c").temp_tag().await?;
264        let dt = blobs.add_slice("d").temp_tag().await?;
265        let et = blobs.add_slice("e").temp_tag().await?;
266        let ft = blobs.add_slice("f").temp_tag().await?;
267        let gt = blobs.add_slice("g").temp_tag().await?;
268        let ht = blobs.add_slice("h").with_named_tag("h").await?;
269        let a = at.hash();
270        let b = bt.hash();
271        let c = ct.hash();
272        let d = dt.hash();
273        let e = et.hash();
274        let f = ft.hash();
275        let g = gt.hash();
276        let h = ht.hash;
277        store.tags().set("c", ct.hash_and_format()).await?;
278        let dehs = [d, e].into_iter().collect::<HashSeq>();
279        let hehs = blobs
280            .add_bytes_with_opts(AddBytesOptions {
281                data: dehs.into(),
282                format: BlobFormat::HashSeq,
283            })
284            .await?;
285        let fghs = [f, g].into_iter().collect::<HashSeq>();
286        let fghs = blobs
287            .add_bytes_with_opts(AddBytesOptions {
288                data: fghs.into(),
289                format: BlobFormat::HashSeq,
290            })
291            .temp_tag()
292            .await?;
293        store.tags().set("fg", fghs.hash_and_format()).await?;
294        drop(fghs);
295        drop(bt);
296        store.tags().delete("h").await?;
297        let mut live = HashSet::new();
298        gc_run_once(store, &mut live).await?;
299        // a is protected because we keep the temp tag
300        assert!(live.contains(&a));
301        assert!(store.has(a).await?);
302        // b is not protected because we drop the temp tag
303        assert!(!live.contains(&b));
304        assert!(!store.has(b).await?);
305        // c is protected because we set an explicit tag
306        assert!(live.contains(&c));
307        assert!(store.has(c).await?);
308        // d and e are protected because they are part of a hashseq protected by a temp tag
309        assert!(live.contains(&d));
310        assert!(store.has(d).await?);
311        assert!(live.contains(&e));
312        assert!(store.has(e).await?);
313        // f and g are protected because they are part of a hashseq protected by a tag
314        assert!(live.contains(&f));
315        assert!(store.has(f).await?);
316        assert!(live.contains(&g));
317        assert!(store.has(g).await?);
318        // h is not protected because we deleted the tag before gc ran
319        assert!(!live.contains(&h));
320        assert!(!store.has(h).await?);
321        drop(at);
322        drop(hehs);
323        Ok(())
324    }
325
326    #[cfg(feature = "fs-store")]
327    async fn gc_file_delete(path: &std::path::Path, store: &Store) -> TestResult<()> {
328        use bao_tree::ChunkNum;
329
330        use crate::store::{fs::options::PathOptions, util::tests::create_n0_bao};
331        let mut live = HashSet::new();
332        let options = PathOptions::new(&path.join("db"));
333        // create a large complete file and check that the data and outboard files are deleted by gc
334        {
335            let a = store
336                .blobs()
337                .add_slice(vec![0u8; 8000000])
338                .temp_tag()
339                .await?;
340            let ah = a.hash();
341            let data_path = options.data_path(&ah);
342            let outboard_path = options.outboard_path(&ah);
343            assert!(data_path.exists());
344            assert!(outboard_path.exists());
345            assert!(store.has(ah).await?);
346            drop(a);
347            gc_run_once(store, &mut live).await?;
348            assert!(!data_path.exists());
349            assert!(!outboard_path.exists());
350        }
351        live.clear();
352        // create a large partial file and check that the data and outboard file as well as
353        // the sizes and bitfield files are deleted by gc
354        {
355            let data = vec![1u8; 8000000];
356            let ranges = ChunkRanges::from(..ChunkNum(19));
357            let (bh, b_bao) = create_n0_bao(&data, &ranges)?;
358            store.import_bao_bytes(bh, ranges, b_bao).await?;
359            let data_path = options.data_path(&bh);
360            let outboard_path = options.outboard_path(&bh);
361            let sizes_path = options.sizes_path(&bh);
362            let bitfield_path = options.bitfield_path(&bh);
363            store.wait_idle().await?;
364            assert!(data_path.exists());
365            assert!(outboard_path.exists());
366            assert!(sizes_path.exists());
367            assert!(bitfield_path.exists());
368            gc_run_once(store, &mut live).await?;
369            assert!(!data_path.exists());
370            assert!(!outboard_path.exists());
371            assert!(!sizes_path.exists());
372            assert!(!bitfield_path.exists());
373        }
374        Ok(())
375    }
376
377    #[tokio::test]
378    #[cfg(feature = "fs-store")]
379    async fn gc_smoke_fs() -> TestResult {
380        tracing_subscriber::fmt::try_init().ok();
381        let testdir = tempfile::tempdir()?;
382        let db_path = testdir.path().join("db");
383        let store = crate::store::fs::FsStore::load(&db_path).await?;
384        gc_smoke(&store).await?;
385        gc_file_delete(testdir.path(), &store).await?;
386        Ok(())
387    }
388
389    #[tokio::test]
390    async fn gc_smoke_mem() -> TestResult {
391        tracing_subscriber::fmt::try_init().ok();
392        let store = crate::store::mem::MemStore::new();
393        gc_smoke(&store).await?;
394        Ok(())
395    }
396
397    #[tokio::test]
398    #[cfg(feature = "fs-store")]
399    async fn gc_check_deletion_fs() -> TestResult {
400        tracing_subscriber::fmt::try_init().ok();
401        let testdir = tempfile::tempdir()?;
402        let db_path = testdir.path().join("db");
403        let store = crate::store::fs::FsStore::load(&db_path).await?;
404        gc_check_deletion(&store).await
405    }
406
407    #[tokio::test]
408    async fn gc_check_deletion_mem() -> TestResult {
409        tracing_subscriber::fmt::try_init().ok();
410        let store = crate::store::mem::MemStore::default();
411        gc_check_deletion(&store).await
412    }
413
414    async fn gc_check_deletion(store: &Store) -> TestResult {
415        let temp_tag = store.add_bytes(b"foo".to_vec()).temp_tag().await?;
416        let hash = temp_tag.hash();
417        assert_eq!(store.get_bytes(hash).await?.as_ref(), b"foo");
418        drop(temp_tag);
419        let mut live = HashSet::new();
420        gc_run_once(store, &mut live).await?;
421
422        // check that `get_bytes` returns an error.
423        let res = store.get_bytes(hash).await;
424        assert!(res.is_err());
425        assert!(matches!(
426            res,
427            Err(ExportBaoError::ExportBaoInner {
428                source: EncodeError::Io(cause),
429                ..
430            }) if cause.kind() == io::ErrorKind::NotFound
431        ));
432
433        // check that `export_ranges` returns an error.
434        let res = store
435            .export_ranges(hash, RangeSet2::all())
436            .concatenate()
437            .await;
438        assert!(res.is_err());
439        assert!(matches!(
440            res,
441            Err(RequestError::Inner{
442                source: crate::api::Error::Io(cause),
443                ..
444            }) if cause.kind() == io::ErrorKind::NotFound
445        ));
446
447        // check that `export_bao` returns an error.
448        let res = store
449            .export_bao(hash, ChunkRanges::all())
450            .bao_to_vec()
451            .await;
452        assert!(res.is_err());
453        println!("export_bao res {res:?}");
454        assert!(matches!(
455            res,
456            Err(RequestError::Inner{
457                source: crate::api::Error::Io(cause),
458                ..
459            }) if cause.kind() == io::ErrorKind::NotFound
460        ));
461
462        // check that `export` returns an error.
463        let target = tempfile::NamedTempFile::new()?;
464        let path = target.path();
465        let res = store.export(hash, path).await;
466        assert!(res.is_err());
467        assert!(matches!(
468            res,
469            Err(RequestError::Inner{
470                source: crate::api::Error::Io(cause),
471                ..
472            }) if cause.kind() == io::ErrorKind::NotFound
473        ));
474        Ok(())
475    }
476}
iroh_blobs/store/gc.rs

iroh_blobs/store/
gc.rs
