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