Skip to main content

max / goingson

5.3 KB · 140 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(blobs_dir: &Path, referenced: &HashSet<String>) -> std::io::Result<usize> {
59 if !blobs_dir.exists() {
60 return Ok(0);
61 }
62
63 let mut removed = 0;
64 for entry in std::fs::read_dir(blobs_dir)? {
65 let entry = entry?;
66 if !entry.file_type()?.is_file() {
67 continue;
68 }
69 let name = entry.file_name();
70 let Some(name) = name.to_str() else { continue };
71 if referenced.contains(name) {
72 continue;
73 }
74 match std::fs::remove_file(entry.path()) {
75 Ok(()) => removed += 1,
76 Err(e) => warn!("Blob GC: failed to remove orphan {name}: {e}"),
77 }
78 }
79 Ok(removed)
80 }
81
82 /// Reconcile the blob store against live references and unlink orphans.
83 ///
84 /// Best-effort: any failure is logged and swallowed so startup is never blocked.
85 pub async fn reconcile(pool: &SqlitePool, data_dir: &Path) {
86 let referenced = match collect_referenced_hashes(pool).await {
87 Ok(set) => set,
88 Err(e) => {
89 warn!("Blob GC skipped: could not collect references: {e}");
90 return;
91 }
92 };
93
94 // read_dir + remove_file are blocking; run on the blocking pool so the
95 // startup GC doesn't stall the reactor / delay first paint (Perf M4).
96 let blobs_dir = data_dir.join("blobs");
97 match tokio::task::spawn_blocking(move || reclaim_orphans(&blobs_dir, &referenced)).await {
98 Ok(Ok(0)) => {}
99 Ok(Ok(n)) => info!("Blob GC reclaimed {n} orphaned blob(s)"),
100 Ok(Err(e)) => warn!("Blob GC: could not scan blob dir: {e}"),
101 Err(e) => warn!("Blob GC task panicked: {e}"),
102 }
103 }
104
105 #[cfg(test)]
106 mod tests {
107 use super::*;
108
109 #[test]
110 fn reclaim_removes_only_unreferenced_files() {
111 let dir = std::env::temp_dir().join(format!("goingson-blobgc-test-{}", std::process::id()));
112 let blobs = dir.join("blobs");
113 std::fs::create_dir_all(&blobs).unwrap();
114
115 std::fs::write(blobs.join("keep_me"), b"a").unwrap();
116 std::fs::write(blobs.join("orphan_1"), b"b").unwrap();
117 std::fs::write(blobs.join("orphan_2"), b"c").unwrap();
118
119 let mut referenced = HashSet::new();
120 referenced.insert("keep_me".to_string());
121 // A referenced-but-missing hash must not error (blob simply not local yet).
122 referenced.insert("not_on_disk".to_string());
123
124 let removed = reclaim_orphans(&blobs, &referenced).unwrap();
125 assert_eq!(removed, 2);
126 assert!(blobs.join("keep_me").exists());
127 assert!(!blobs.join("orphan_1").exists());
128 assert!(!blobs.join("orphan_2").exists());
129
130 std::fs::remove_dir_all(&dir).ok();
131 }
132
133 #[test]
134 fn reclaim_on_missing_dir_is_noop() {
135 let missing = std::env::temp_dir().join("goingson-blobgc-does-not-exist-xyz");
136 let removed = reclaim_orphans(&missing, &HashSet::new()).unwrap();
137 assert_eq!(removed, 0);
138 }
139 }
140