didbot_serve/
blob_collect.rs

1//! The timer that runs `didbot_pds::Provisioner::collect_blobs`.
2//!
3//! An upload starts unreferenced, and a blob whose reference count has
4//! fallen to zero stays on disk until something unlinks it. `collect_blobs`
5//! is that something, and this is its clock — the same shape as
6//! [`crate::stale_sweep`], for the same reason: a startup-only pass would
7//! leave a long-lived server accumulating bytes nothing names.
8
9use std::panic::AssertUnwindSafe;
10use std::sync::Arc;
11use std::time::Duration;
12
13use didbot_pds::BlobCollect;
14use tokio::sync::Notify;
15use tokio::task::JoinHandle;
16use tokio::time::MissedTickBehavior;
17use tracing::{info, warn};
18
19/// How often the collector runs, absent configuration.
20///
21/// An hour: the floor under a deployment's grace window is
22/// `didbot_pds::blobs::MIN_BLOB_COLLECTION_GRACE`, an hour, so a blob past
23/// its window waits at most another window's worth before a pass takes it.
24pub const DEFAULT_COLLECT_INTERVAL: Duration = Duration::from_secs(60 * 60);
25
26/// A running collector, stopped by [`BlobCollector::stop`] or by dropping it.
27pub struct BlobCollector {
28    handle: JoinHandle<()>,
29    stop: Arc<Notify>,
30}
31
32impl BlobCollector {
33    /// Starts collecting through `registry` every `interval`.
34    ///
35    /// The first tick fires immediately, so a restart takes what the
36    /// previous run left past its grace without waiting an interval. That
37    /// pass runs alongside serving, like every later one.
38    ///
39    /// Each pass runs on a blocking thread, as [`crate::stale_sweep`]'s does:
40    /// it takes the write lock every write path takes and unlinks files.
41    pub fn spawn(registry: Arc<dyn BlobCollect>, interval: Duration) -> Self {
42        let stop = Arc::new(Notify::new());
43        let notified = stop.clone();
44        let handle = tokio::spawn(async move {
45            let mut ticker = tokio::time::interval(interval);
46            ticker.set_missed_tick_behavior(MissedTickBehavior::Delay);
47            loop {
48                tokio::select! {
49                    biased;
50                    () = notified.notified() => break,
51                    _ = ticker.tick() => {}
52                }
53                collect_once(registry.clone()).await;
54            }
55        });
56        Self { handle, stop }
57    }
58
59    /// Stops the collector and waits for the task to finish.
60    pub async fn stop(mut self) {
61        self.stop.notify_one();
62        let _ = (&mut self.handle).await;
63    }
64}
65
66impl Drop for BlobCollector {
67    fn drop(&mut self) {
68        self.handle.abort();
69    }
70}
71
72/// One tick: collect, say what it took.
73///
74/// Public and separate from the timer for the reason
75/// [`crate::stale_sweep::sweep_once`] is: a caller can run exactly one pass
76/// without waiting on a clock.
77pub async fn collect_once(registry: Arc<dyn BlobCollect>) {
78    let collected = tokio::task::spawn_blocking(move || {
79        std::panic::catch_unwind(AssertUnwindSafe(|| registry.collect_blobs()))
80    })
81    .await;
82
83    match collected {
84        Ok(Ok(collected)) if collected.is_empty() => {}
85        Ok(Ok(collected)) => info!(
86            blobs = collected.len(),
87            bytes = collected.iter().map(|blob| blob.bytes).sum::<u64>(),
88            "collected blobs nothing referenced, past their grace window"
89        ),
90        // Both arms leave the timer running: what this pass did not take
91        // is still there for the next one.
92        Ok(Err(_)) => warn!("the blob collector panicked; the next tick tries again"),
93        Err(error) => {
94            warn!(%error, "the blob collector did not finish; the next tick tries again")
95        }
96    }
97}
didbot_serve/blob_collect.rs

didbot_serve/
blob_collect.rs
