Skip to main content

max / goingson

5.3 KB · 143 lines History Blame Raw
1 //! Local blob-store garbage collection.
2 //!
3 //! The content-addressed blob store under `<data_dir>/blobs` accumulates files
4 //! that lose every reference when attachments go away, either through
5 //! `delete_attachment`, or through the `ON DELETE CASCADE` on
6 //! `attachments.task_id` / `attachments.project_id`, which removes attachment
7 //! rows without ever calling `delete_attachment`. Left alone the orphaned blobs
8 //! are a permanent disk leak.
9 //!
10 //! [`reconcile`] walks the blob directory once at startup and unlinks any file
11 //! that nothing references. References come from two places, and BOTH must be
12 //! consulted, deleting a blob an email still points at would be data loss:
13 //! 1. `attachments.blob_hash`, task / project attachments.
14 //! 2. `emails.attachment_meta`, email attachments not yet converted to rows.
15 //!
16 //! Running at startup (before the sync/email schedulers spawn and before the UI
17 //! can create attachments) keeps it free of races with concurrent blob writers.
18
19 use std::collections::HashSet;
20 use std::path::Path;
21
22 use goingson_core::AttachmentMeta;
23 use sqlx::SqlitePool;
24 use tracing::{info, warn};
25
26 /// Collect every blob hash currently referenced by an attachment row or an
27 /// email's `attachment_meta` JSON.
28 pub async fn collect_referenced_hashes(pool: &SqlitePool) -> Result<HashSet<String>, sqlx::Error> {
29 let mut referenced = HashSet::new();
30
31 let attachment_hashes: Vec<(String,)> =
32 sqlx::query_as("SELECT DISTINCT blob_hash FROM attachments")
33 .fetch_all(pool)
34 .await?;
35 referenced.extend(attachment_hashes.into_iter().map(|(h,)| h));
36
37 let metas: Vec<(Option<String>,)> = sqlx::query_as(
38 "SELECT attachment_meta FROM emails WHERE attachment_meta IS NOT NULL AND attachment_meta != ''",
39 )
40 .fetch_all(pool)
41 .await?;
42 for (meta,) in metas {
43 let Some(json) = meta else { continue };
44 match serde_json::from_str::<Vec<AttachmentMeta>>(&json) {
45 Ok(list) => referenced.extend(list.into_iter().map(|m| m.blob_hash)),
46 // A malformed row must never widen the delete set: keep its blobs.
47 Err(e) => warn!("Blob GC: skipping unparseable attachment_meta: {e}"),
48 }
49 }
50
51 Ok(referenced)
52 }
53
54 /// Remove blob files in `blobs_dir` whose name is not in `referenced`.
55 ///
56 /// Returns the number of files unlinked. Non-files and unreadable entries are
57 /// skipped; a failed unlink is logged and does not abort the sweep.
58 pub fn reclaim_orphans<S: std::hash::BuildHasher>(
59 blobs_dir: &Path,
60 referenced: &HashSet<String, S>,
61 ) -> std::io::Result<usize> {
62 if !blobs_dir.exists() {
63 return Ok(0);
64 }
65
66 let mut removed = 0;
67 for entry in std::fs::read_dir(blobs_dir)? {
68 let entry = entry?;
69 if !entry.file_type()?.is_file() {
70 continue;
71 }
72 let name = entry.file_name();
73 let Some(name) = name.to_str() else { continue };
74 if referenced.contains(name) {
75 continue;
76 }
77 match std::fs::remove_file(entry.path()) {
78 Ok(()) => removed += 1,
79 Err(e) => warn!("Blob GC: failed to remove orphan {name}: {e}"),
80 }
81 }
82 Ok(removed)
83 }
84
85 /// Reconcile the blob store against live references and unlink orphans.
86 ///
87 /// Best-effort: any failure is logged and swallowed so startup is never blocked.
88 pub async fn reconcile(pool: &SqlitePool, data_dir: &Path) {
89 let referenced = match collect_referenced_hashes(pool).await {
90 Ok(set) => set,
91 Err(e) => {
92 warn!("Blob GC skipped: could not collect references: {e}");
93 return;
94 }
95 };
96
97 // read_dir + remove_file are blocking; run on the blocking pool so the
98 // startup GC doesn't stall the reactor / delay first paint (Perf M4).
99 let blobs_dir = data_dir.join("blobs");
100 match tokio::task::spawn_blocking(move || reclaim_orphans(&blobs_dir, &referenced)).await {
101 Ok(Ok(0)) => {}
102 Ok(Ok(n)) => info!("Blob GC reclaimed {n} orphaned blob(s)"),
103 Ok(Err(e)) => warn!("Blob GC: could not scan blob dir: {e}"),
104 Err(e) => warn!("Blob GC task panicked: {e}"),
105 }
106 }
107
108 #[cfg(test)]
109 mod tests {
110 use super::*;
111
112 #[test]
113 fn reclaim_removes_only_unreferenced_files() {
114 let dir = std::env::temp_dir().join(format!("goingson-blobgc-test-{}", std::process::id()));
115 let blobs = dir.join("blobs");
116 std::fs::create_dir_all(&blobs).unwrap();
117
118 std::fs::write(blobs.join("keep_me"), b"a").unwrap();
119 std::fs::write(blobs.join("orphan_1"), b"b").unwrap();
120 std::fs::write(blobs.join("orphan_2"), b"c").unwrap();
121
122 let mut referenced = HashSet::new();
123 referenced.insert("keep_me".to_string());
124 // A referenced-but-missing hash must not error (blob not local yet).
125 referenced.insert("not_on_disk".to_string());
126
127 let removed = reclaim_orphans(&blobs, &referenced).unwrap();
128 assert_eq!(removed, 2);
129 assert!(blobs.join("keep_me").exists());
130 assert!(!blobs.join("orphan_1").exists());
131 assert!(!blobs.join("orphan_2").exists());
132
133 std::fs::remove_dir_all(&dir).ok();
134 }
135
136 #[test]
137 fn reclaim_on_missing_dir_is_noop() {
138 let missing = std::env::temp_dir().join("goingson-blobgc-does-not-exist-xyz");
139 let removed = reclaim_orphans(&missing, &HashSet::new()).unwrap();
140 assert_eq!(removed, 0);
141 }
142 }
143