1use std::collections::HashSet;
2use std::fs;
3use std::path::Path;
4use std::time::Duration;
5
6use anyhow::Result;
7use chrono::{DateTime, Utc};
8use sqlx::{Row, SqliteConnection};
9
10use crate::attachments::storage::object_path;
11use crate::attachments::validation::validate_sha256;
12use crate::db::begin_immediate;
13
14use super::filesystem::{
15 FileMove, move_file_without_replacing, remove_files, restore_trashed_files, scan_directory_all,
16 scan_directory_page,
17};
18use super::liveness::{is_protected, live_blob_references_sql, reconcile_liveness_bounded};
19use super::{
20 ByteCount, Clock, LEASE_TTL, LifecyclePolicy, PruneSummary, cutoff, staging_dir, timestamp,
21 trash_dir,
22};
23
24async fn reconcile_trash_files(
25 conn: &mut SqliteConnection,
26 blob_dir: &Path,
27 files: Vec<super::filesystem::ScannedFile>,
28) -> Result<()> {
29 let hashes = files
30 .iter()
31 .filter(|file| validate_sha256(&file.name).is_ok())
32 .map(|file| file.name.clone())
33 .collect::<Vec<_>>();
34 let available = if hashes.is_empty() {
35 HashSet::new()
36 } else {
37 sqlx::query_scalar::<_, String>(
38 "SELECT sha256 FROM blob_inventory
39 WHERE available = 1 AND sha256 IN (SELECT value FROM json_each(?))",
40 )
41 .bind(serde_json::to_string(&hashes)?)
42 .fetch_all(&mut *conn)
43 .await?
44 .into_iter()
45 .collect()
46 };
47 let blob_dir = blob_dir.to_path_buf();
48 crate::attachments::blocking::run(move || {
49 for file in files {
50 if validate_sha256(&file.name).is_err() {
51 continue;
52 }
53 let target = object_path(&blob_dir, &file.name)?;
54 if available.contains(&file.name) {
55 if target.exists() {
56 fs::remove_file(file.path)?;
57 } else {
58 if let Some(parent) = target.parent() {
59 fs::create_dir_all(parent)?;
60 }
61 match move_file_without_replacing(&file.path, &target)? {
62 FileMove::Moved | FileMove::SourceMissing | FileMove::TargetExists => {}
63 }
64 }
65 } else {
66 match fs::remove_file(file.path) {
67 Ok(()) => {}
68 Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
69 Err(error) => return Err(error.into()),
70 }
71 }
72 }
73 Ok(())
74 })
75 .await
76}
77
78pub async fn reconcile_trash(conn: &mut SqliteConnection, blob_dir: &Path) -> Result<()> {
79 reconcile_trash_files(
80 conn,
81 blob_dir,
82 scan_directory_all(trash_dir(blob_dir)).await?,
83 )
84 .await
85}
86
87async fn reconcile_trash_page(
88 conn: &mut SqliteConnection,
89 blob_dir: &Path,
90 limit: usize,
91) -> Result<()> {
92 reconcile_trash_files(
93 conn,
94 blob_dir,
95 scan_directory_page(trash_dir(blob_dir), limit).await?,
96 )
97 .await
98}
99
100pub async fn reconcile_staging(blob_dir: &Path) -> Result<ByteCount> {
101 let files = scan_directory_all(staging_dir(blob_dir)).await?;
102 let stale = files
103 .into_iter()
104 .filter(|file| {
105 file.name.starts_with(".aven-stage-")
106 && file.modified.elapsed().is_ok_and(|age| age >= LEASE_TTL)
107 })
108 .collect::<Vec<_>>();
109 let removed = ByteCount {
110 count: u64::try_from(stale.len())?,
111 bytes: stale.iter().map(|file| file.len).sum(),
112 };
113 remove_files(stale.into_iter().map(|file| file.path).collect()).await?;
114 Ok(removed)
115}
116
117pub async fn reconcile_missing_objects(
118 conn: &mut SqliteConnection,
119 blob_dir: &Path,
120 clock: &dyn Clock,
121) -> Result<ByteCount> {
122 let rows = sqlx::query(
123 "SELECT sha256, byte_size FROM blob_inventory
124 WHERE available = 1 ORDER BY sha256",
125 )
126 .fetch_all(&mut *conn)
127 .await?;
128 let verified_at = timestamp(clock.now());
129 let mut missing = ByteCount::default();
130 for row in rows {
131 let sha256: String = row.get("sha256");
132 if object_path(blob_dir, &sha256)?.exists() {
133 continue;
134 }
135 sqlx::query(
136 "UPDATE blob_inventory SET available = 0, last_verified_at = ?
137 WHERE sha256 = ? AND available = 1",
138 )
139 .bind(&verified_at)
140 .bind(&sha256)
141 .execute(&mut *conn)
142 .await?;
143 missing.count += 1;
144 missing.bytes += u64::try_from(row.get::<i64, _>("byte_size"))?;
145 }
146 Ok(missing)
147}
148
149async fn reconcile_object_directory(
150 conn: &mut SqliteConnection,
151 blob_dir: &Path,
152 grace: Duration,
153 limit: usize,
154 clock: &dyn Clock,
155) -> Result<ByteCount> {
156 let files = scan_directory_page(staging_dir(blob_dir), limit).await?;
157 let stale_staging = files
158 .iter()
159 .filter(|file| {
160 file.name.starts_with(".aven-stage-")
161 && file.modified.elapsed().is_ok_and(|age| age >= LEASE_TTL)
162 })
163 .map(|file| file.path.clone())
164 .collect::<Vec<_>>();
165 remove_files(stale_staging).await?;
166
167 let cutoff = clock.now() - chrono::Duration::from_std(grace)?;
168 let canonical = files
169 .iter()
170 .filter(|file| {
171 validate_sha256(&file.name).is_ok() && DateTime::<Utc>::from(file.modified) <= cutoff
172 })
173 .collect::<Vec<_>>();
174 if canonical.is_empty() {
175 return Ok(ByteCount::default());
176 }
177 let hashes = canonical
178 .iter()
179 .map(|file| file.name.clone())
180 .collect::<Vec<_>>();
181 let candidates = serde_json::to_string(&hashes)?;
182 let now = timestamp(clock.now());
183 let live_blob_references = live_blob_references_sql("candidate.value");
184 let protected = sqlx::query_scalar::<_, String>(sqlx::AssertSqlSafe(format!(
185 "SELECT candidate.value FROM json_each(?) candidate
186 WHERE EXISTS(SELECT 1 FROM blob_inventory bi WHERE bi.sha256 = candidate.value)
187 OR {live_blob_references}
188 OR EXISTS(
189 SELECT 1 FROM changes
190 WHERE server_seq IS NULL AND op_type = 'attachment_add'
191 AND json_extract(payload, '$.sha256') = candidate.value
192 )
193 OR EXISTS(
194 SELECT 1 FROM blob_leases
195 WHERE sha256 = candidate.value AND expires_at > ?
196 )
197 OR EXISTS(
198 SELECT 1 FROM blob_upload_reservations
199 WHERE sha256 = candidate.value AND expires_at > ?
200 )"
201 )))
202 .bind(candidates)
203 .bind(&now)
204 .bind(&now)
205 .fetch_all(&mut *conn)
206 .await?
207 .into_iter()
208 .collect::<HashSet<_>>();
209 let removable = canonical
210 .into_iter()
211 .filter(|file| !protected.contains(&file.name))
212 .map(|file| (file.path.clone(), file.name.clone(), file.len))
213 .collect::<Vec<_>>();
214 if removable.is_empty() {
215 return Ok(ByteCount::default());
216 }
217 let blob_dir = blob_dir.to_path_buf();
218 crate::attachments::blocking::run(move || {
219 let trash = trash_dir(&blob_dir);
220 fs::create_dir_all(&trash)?;
221 let mut removed = ByteCount::default();
222 for (source, name, len) in removable {
223 let target = trash.join(name);
224 if move_file_without_replacing(&source, &target)? == FileMove::Moved {
225 fs::remove_file(target)?;
226 removed.count += 1;
227 removed.bytes += len;
228 }
229 }
230 Ok(removed)
231 })
232 .await
233}
234
235pub async fn reconcile_orphan_objects(
236 conn: &mut SqliteConnection,
237 blob_dir: &Path,
238 grace: Duration,
239 clock: &dyn Clock,
240) -> Result<ByteCount> {
241 reconcile_object_directory(conn, blob_dir, grace, usize::MAX, clock).await
242}
243
244pub async fn prune(
245 conn: &mut SqliteConnection,
246 blob_dir: &Path,
247 policy: LifecyclePolicy,
248 apply: bool,
249 clock: &dyn Clock,
250) -> Result<PruneSummary> {
251 if apply {
252 reconcile_trash_page(conn, blob_dir, policy.maintenance_limit).await?;
253 reconcile_object_directory(
254 conn,
255 blob_dir,
256 policy.grace,
257 policy.maintenance_limit,
258 clock,
259 )
260 .await?;
261 reconcile_missing_objects(conn, blob_dir, clock).await?;
262 }
263 reconcile_liveness_bounded(conn, policy.maintenance_limit, clock).await?;
264 let now = timestamp(clock.now());
265 let cutoff = cutoff(clock.now(), policy.grace)?;
266 let rows = sqlx::query(
267 "SELECT bi.sha256, bi.byte_size
268 FROM blob_inventory bi
269 JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256
270 WHERE bi.available = 1 AND bl.unreferenced_at IS NOT NULL
271 AND bl.unreferenced_at <= ?
272 ORDER BY bl.unreferenced_at, bi.sha256 LIMIT ?",
273 )
274 .bind(&cutoff)
275 .bind(i64::try_from(policy.maintenance_limit)?)
276 .fetch_all(&mut *conn)
277 .await?;
278 let mut summary = PruneSummary::default();
279 for row in rows {
280 let sha256: String = row.get("sha256");
281 let byte_size: i64 = row.get("byte_size");
282 if is_protected(conn, &sha256, &now).await? {
283 continue;
284 }
285 summary.eligible.count += 1;
286 summary.eligible.bytes += u64::try_from(byte_size)?;
287 if !apply {
288 continue;
289 }
290 let mut tx = begin_immediate(conn).await?;
291 let still_eligible: bool = sqlx::query_scalar(
292 "SELECT EXISTS(
293 SELECT 1 FROM blob_inventory bi
294 JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256
295 WHERE bi.sha256 = ? AND bi.available = 1
296 AND bl.unreferenced_at IS NOT NULL AND bl.unreferenced_at <= ?
297 )",
298 )
299 .bind(&sha256)
300 .bind(&cutoff)
301 .fetch_one(&mut *tx)
302 .await?;
303 if !still_eligible || is_protected(&mut tx, &sha256, &now).await? {
304 tx.rollback().await?;
305 continue;
306 }
307 let source = object_path(blob_dir, &sha256)?;
308 let trash = trash_dir(blob_dir);
309 let trashed = trash.join(&sha256);
310 let source_for_move = source.clone();
311 let trashed_for_move = trashed.clone();
312 let moved = crate::attachments::blocking::run(move || {
313 fs::create_dir_all(&trash)?;
314 move_file_without_replacing(&source_for_move, &trashed_for_move)
315 })
316 .await?;
317 if moved == FileMove::TargetExists {
318 tx.rollback().await?;
319 continue;
320 }
321 if let Err(error) = sqlx::query(
322 "UPDATE blob_inventory SET available = 0, last_verified_at = ? WHERE sha256 = ?",
323 )
324 .bind(&now)
325 .bind(&sha256)
326 .execute(&mut *tx)
327 .await
328 {
329 let _ = tx.rollback().await;
330 if moved == FileMove::Moved {
331 restore_trashed_files(vec![(source, trashed)]).await;
332 }
333 return Err(error.into());
334 }
335 if let Err(error) = tx.commit().await {
336 if moved == FileMove::Moved {
337 restore_trashed_files(vec![(source, trashed)]).await;
338 }
339 return Err(error.into());
340 }
341 if moved == FileMove::Moved {
342 remove_files(vec![trashed]).await?;
343 }
344 summary.pruned.count += 1;
345 summary.pruned.bytes += u64::try_from(byte_size)?;
346 }
347 let blob_dir = blob_dir.to_path_buf();
348 crate::attachments::blocking::run(move || {
349 prune_preview_cache(&blob_dir, policy.preview_quota_bytes)
350 })
351 .await?;
352 Ok(summary)
353}
354
355pub fn prune_preview_cache(blob_dir: &Path, quota: u64) -> Result<ByteCount> {
356 let root = blob_dir.join("cache").join("previews");
357 if !root.exists() {
358 return Ok(ByteCount::default());
359 }
360 let mut files = Vec::new();
361 let mut dirs = vec![root];
362 while let Some(dir) = dirs.pop() {
363 for entry in fs::read_dir(dir)? {
364 let entry = entry?;
365 if entry.file_type()?.is_dir() {
366 dirs.push(entry.path());
367 } else if entry.file_type()?.is_file() {
368 let metadata = entry.metadata()?;
369 files.push((metadata.modified()?, metadata.len(), entry.path()));
370 }
371 }
372 }
373 let mut total: u64 = files.iter().map(|(_, size, _)| *size).sum();
374 files.sort_by_key(|(modified, _, path)| (*modified, path.clone()));
375 let mut removed = ByteCount::default();
376 for (_, size, path) in files {
377 if total <= quota {
378 break;
379 }
380 fs::remove_file(path)?;
381 total -= size;
382 removed.count += 1;
383 removed.bytes += size;
384 }
385 Ok(removed)
386}
387
388#[cfg(test)]
389mod tests {
390 use std::fs;
391 use std::time::Duration;
392
393 use sqlx::Connection as _;
394 use sqlx::SqliteConnection;
395 use sqlx::sqlite::SqliteConnectOptions;
396
397 use super::super::test_support::{TestClock, insert_attachment, insert_task};
398 use super::super::*;
399 use super::*;
400 use crate::attachments::storage::{object_path, upsert_inventory_available};
401 use crate::db::{begin_immediate, open_db};
402
403 #[tokio::test]
404 async fn concurrent_attach_wins_prune_recheck() {
405 let temp = tempfile::tempdir().unwrap();
406 let db_path = temp.path().join("test.sqlite");
407 let pool = open_db(&db_path).await.unwrap();
408 let mut conn = pool.acquire().await.unwrap();
409 let blob_dir = temp.path().join("blobs");
410 let hash = "abababababababababababababababababababababababababababababababab";
411 upsert_inventory_available(&mut conn, hash, 4, "image/png")
412 .await
413 .unwrap();
414 insert_task(&mut conn, "0000000000000003").await;
415 let path = object_path(&blob_dir, hash).unwrap();
416 fs::create_dir_all(path.parent().unwrap()).unwrap();
417 fs::write(&path, b"blob").unwrap();
418 let clock = TestClock::at("2026-07-01T00:00:00Z");
419 reconcile_liveness(&mut conn, &clock).await.unwrap();
420 clock.advance(chrono::Duration::days(8));
421 let options = SqliteConnectOptions::new()
422 .filename(&db_path)
423 .busy_timeout(Duration::from_secs(5));
424 let prune_conn = SqliteConnection::connect_with(&options).await.unwrap();
425
426 let mut tx = begin_immediate(&mut conn).await.unwrap();
427 insert_attachment(&mut tx, "0000000000000013", "0000000000000003", hash, false).await;
428 let prune_dir = blob_dir.clone();
429 let prune_clock = clock.clone();
430 let pruning = tokio::spawn(async move {
431 let mut prune_conn = prune_conn;
432 prune(
433 &mut prune_conn,
434 &prune_dir,
435 LifecyclePolicy::default(),
436 true,
437 &prune_clock,
438 )
439 .await
440 .unwrap()
441 });
442 tokio::time::sleep(Duration::from_millis(50)).await;
443 tx.commit().await.unwrap();
444
445 let summary = pruning.await.unwrap();
446 assert_eq!(summary.pruned.count, 0);
447 assert!(path.exists());
448 }
449
450 #[tokio::test]
451 async fn interrupted_atomic_create_object_is_reconciled_after_grace() {
452 let temp = tempfile::tempdir().unwrap();
453 let db_path = temp.path().join("test.sqlite");
454 let pool = open_db(&db_path).await.unwrap();
455 let mut conn = pool.acquire().await.unwrap();
456 let blob_dir = temp.path().join("blobs");
457 let hash = "cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd";
458 let path = object_path(&blob_dir, hash).unwrap();
459 fs::create_dir_all(path.parent().unwrap()).unwrap();
460 fs::write(&path, b"orphan").unwrap();
461 let clock = TestClock::at("2030-07-01T00:00:00Z");
462 let policy = LifecyclePolicy {
463 grace: Duration::ZERO,
464 ..LifecyclePolicy::default()
465 };
466
467 prune(&mut conn, &blob_dir, policy, true, &clock)
468 .await
469 .unwrap();
470 assert!(!path.exists());
471 }
472
473 #[tokio::test]
474 async fn orphan_traversal_obeys_limit_and_resumes_between_steps() {
475 let temp = tempfile::tempdir().unwrap();
476 let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
477 let mut conn = pool.acquire().await.unwrap();
478 let blob_dir = temp.path().join("blobs");
479 for digit in ['6', '7', '8', '9', 'a'] {
480 let hash = digit.to_string().repeat(64);
481 let path = object_path(&blob_dir, &hash).unwrap();
482 fs::create_dir_all(path.parent().unwrap()).unwrap();
483 fs::write(path, b"orphan").unwrap();
484 }
485 let clock = TestClock::at("2030-07-01T00:00:00Z");
486 let policy = LifecyclePolicy {
487 grace: Duration::ZERO,
488 maintenance_limit: 2,
489 ..LifecyclePolicy::default()
490 };
491
492 prune(&mut conn, &blob_dir, policy, true, &clock)
493 .await
494 .unwrap();
495 let after_one = fs::read_dir(staging_dir(&blob_dir)).unwrap().count();
496 assert!(after_one >= 3);
497
498 for _ in 0..4 {
499 prune(&mut conn, &blob_dir, policy, true, &clock)
500 .await
501 .unwrap();
502 }
503 assert_eq!(fs::read_dir(staging_dir(&blob_dir)).unwrap().count(), 0);
504 }
505
506 #[tokio::test]
507 async fn interrupted_trash_move_restores_available_object() {
508 let temp = tempfile::tempdir().unwrap();
509 let db_path = temp.path().join("test.sqlite");
510 let pool = open_db(&db_path).await.unwrap();
511 let mut conn = pool.acquire().await.unwrap();
512 let blob_dir = temp.path().join("blobs");
513 let hash = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
514 upsert_inventory_available(&mut conn, hash, 4, "image/png")
515 .await
516 .unwrap();
517 let source = object_path(&blob_dir, hash).unwrap();
518 let trash = trash_dir(&blob_dir).join(hash);
519 fs::create_dir_all(source.parent().unwrap()).unwrap();
520 fs::create_dir_all(trash.parent().unwrap()).unwrap();
521 fs::write(&source, b"blob").unwrap();
522 fs::rename(&source, &trash).unwrap();
523
524 reconcile_trash(&mut conn, &blob_dir).await.unwrap();
525 assert!(source.exists());
526 assert!(!trash.exists());
527 }
528}