1#![allow(dead_code)]
2
3use std::fs;
4use std::path::{Path, PathBuf};
5use std::time::Duration;
6
7use anyhow::{Context, Result, bail};
8use chrono::{DateTime, SecondsFormat, Utc};
9use sqlx::{Row, SqliteConnection};
10
11use crate::attachments::storage::object_path;
12use crate::attachments::validation::validate_sha256;
13use crate::db::begin_immediate;
14use crate::ids::new_id;
15
16pub const DEFAULT_LOCAL_GRACE: Duration = Duration::from_secs(7 * 24 * 60 * 60);
17pub const DEFAULT_ORIGINAL_QUOTA_BYTES: i64 = 10 * 1024 * 1024 * 1024;
18pub const DEFAULT_PREVIEW_QUOTA_BYTES: u64 = 512 * 1024 * 1024;
19pub const DEFAULT_MAINTENANCE_LIMIT: usize = 128;
20const LEASE_TTL: Duration = Duration::from_secs(10 * 60);
21
22pub trait Clock: Send + Sync {
23 fn now(&self) -> DateTime<Utc>;
24}
25
26pub struct SystemClock;
27
28impl Clock for SystemClock {
29 fn now(&self) -> DateTime<Utc> {
30 Utc::now()
31 }
32}
33
34#[derive(Debug, Clone, Copy)]
35pub struct LifecyclePolicy {
36 pub grace: Duration,
37 pub quota_bytes: i64,
38 pub preview_quota_bytes: u64,
39 pub maintenance_limit: usize,
40}
41
42impl Default for LifecyclePolicy {
43 fn default() -> Self {
44 Self {
45 grace: DEFAULT_LOCAL_GRACE,
46 quota_bytes: DEFAULT_ORIGINAL_QUOTA_BYTES,
47 preview_quota_bytes: DEFAULT_PREVIEW_QUOTA_BYTES,
48 maintenance_limit: DEFAULT_MAINTENANCE_LIMIT,
49 }
50 }
51}
52
53#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
54pub struct ByteCount {
55 pub count: u64,
56 pub bytes: u64,
57}
58
59#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
60pub struct LifecycleReport {
61 pub referenced: ByteCount,
62 pub protected: ByteCount,
63 pub grace_period: ByteCount,
64 pub eligible: ByteCount,
65 pub staging: ByteCount,
66 pub trash: ByteCount,
67 pub reservations: ByteCount,
68 pub quota: ByteCount,
69 pub inconsistencies: ByteCount,
70}
71
72#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
73pub struct PruneSummary {
74 pub eligible: ByteCount,
75 pub pruned: ByteCount,
76}
77
78fn timestamp(now: DateTime<Utc>) -> String {
79 now.to_rfc3339_opts(SecondsFormat::Secs, true)
80}
81
82fn cutoff(now: DateTime<Utc>, grace: Duration) -> Result<String> {
83 let grace = chrono::Duration::from_std(grace)?;
84 Ok(timestamp(now - grace))
85}
86
87fn trash_dir(blob_dir: &Path) -> PathBuf {
88 blob_dir.join("trash")
89}
90
91fn staging_dir(blob_dir: &Path) -> PathBuf {
92 blob_dir.join("objects").join("sha256")
93}
94
95fn live_blob_references_sql(sha256_expr: &str) -> String {
96 format!(
97 "EXISTS(
98 SELECT 1 FROM task_attachments ta
99 JOIN tasks t ON t.workspace_id = ta.workspace_id AND t.id = ta.task_id
100 WHERE ta.sha256 = {sha256_expr} AND ta.deleted = 0 AND t.deleted = 0
101 ) OR EXISTS(
102 SELECT 1 FROM server_blob_references sbr
103 LEFT JOIN server_task_tombstones st
104 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
105 WHERE sbr.sha256 = {sha256_expr} AND sbr.deleted = 0
106 AND COALESCE(st.deleted, 0) = 0
107 )"
108 )
109}
110
111pub async fn reconcile_liveness(conn: &mut SqliteConnection, clock: &dyn Clock) -> Result<()> {
112 let mut tx = begin_immediate(conn).await?;
113 reconcile_liveness_in_transaction(&mut tx, clock).await?;
114 tx.commit().await?;
115 Ok(())
116}
117
118pub(crate) async fn reconcile_liveness_in_transaction(
119 conn: &mut SqliteConnection,
120 clock: &dyn Clock,
121) -> Result<()> {
122 let now = timestamp(clock.now());
123 sqlx::query("DELETE FROM blob_leases WHERE expires_at <= ?")
124 .bind(&now)
125 .execute(&mut *conn)
126 .await?;
127 sqlx::query("DELETE FROM blob_upload_reservations WHERE expires_at <= ?")
128 .bind(&now)
129 .execute(&mut *conn)
130 .await?;
131 sqlx::query(
132 "INSERT OR IGNORE INTO blob_lifecycle(sha256, unreferenced_at)
133 SELECT sha256, NULL FROM blob_inventory",
134 )
135 .execute(&mut *conn)
136 .await?;
137 let live_blob_references = live_blob_references_sql("blob_lifecycle.sha256");
138 sqlx::query(sqlx::AssertSqlSafe(format!(
139 "UPDATE blob_lifecycle SET unreferenced_at = NULL
140 WHERE {live_blob_references}"
141 )))
142 .execute(&mut *conn)
143 .await?;
144 sqlx::query(sqlx::AssertSqlSafe(format!(
145 "UPDATE blob_lifecycle SET unreferenced_at = ?
146 WHERE unreferenced_at IS NULL
147 AND NOT ({live_blob_references})"
148 )))
149 .bind(&now)
150 .execute(&mut *conn)
151 .await?;
152 Ok(())
153}
154
155async fn is_protected(conn: &mut SqliteConnection, sha256: &str, now: &str) -> Result<bool> {
156 let live_blob_references = live_blob_references_sql("?");
157 Ok(sqlx::query_scalar::<_, bool>(sqlx::AssertSqlSafe(format!(
158 "SELECT
159 {live_blob_references} OR EXISTS(
160 SELECT 1 FROM changes
161 WHERE server_seq IS NULL AND op_type = 'attachment_add'
162 AND json_extract(payload, '$.sha256') = ?
163 ) OR EXISTS(
164 SELECT 1 FROM blob_leases WHERE sha256 = ? AND expires_at > ?
165 ) OR EXISTS(
166 SELECT 1 FROM blob_upload_reservations WHERE sha256 = ? AND expires_at > ?
167 )"
168 )))
169 .bind(sha256)
170 .bind(sha256)
171 .bind(sha256)
172 .bind(sha256)
173 .bind(now)
174 .bind(sha256)
175 .bind(now)
176 .fetch_one(&mut *conn)
177 .await?)
178}
179
180pub async fn acquire_lease(
181 conn: &mut SqliteConnection,
182 sha256: &str,
183 kind: &str,
184 clock: &dyn Clock,
185) -> Result<String> {
186 validate_sha256(sha256)?;
187 if !matches!(kind, "staging" | "read" | "backup" | "transfer") {
188 bail!("error attachment-lease-kind-invalid");
189 }
190 let lease_id = new_id();
191 let now = clock.now();
192 let expires = now + chrono::Duration::from_std(LEASE_TTL)?;
193 sqlx::query(
194 "INSERT INTO blob_leases(lease_id, sha256, kind, created_at, expires_at)
195 VALUES (?, ?, ?, ?, ?)",
196 )
197 .bind(&lease_id)
198 .bind(sha256)
199 .bind(kind)
200 .bind(timestamp(now))
201 .bind(timestamp(expires))
202 .execute(&mut *conn)
203 .await?;
204 Ok(lease_id)
205}
206
207pub async fn release_lease(conn: &mut SqliteConnection, lease_id: &str) -> Result<()> {
208 sqlx::query("DELETE FROM blob_leases WHERE lease_id = ?")
209 .bind(lease_id)
210 .execute(&mut *conn)
211 .await?;
212 Ok(())
213}
214
215pub async fn reserve_upload(
216 conn: &mut SqliteConnection,
217 workspace_id: &str,
218 sha256: &str,
219 byte_size: i64,
220 quota_bytes: i64,
221 clock: &dyn Clock,
222) -> Result<Option<String>> {
223 validate_sha256(sha256)?;
224 let mut tx = begin_immediate(conn).await?;
225 let existing: bool = sqlx::query_scalar(
226 "SELECT EXISTS(
227 SELECT 1 FROM server_blob_references sbr
228 LEFT JOIN server_task_tombstones st
229 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
230 WHERE sbr.workspace_id = ? AND sbr.sha256 = ? AND sbr.deleted = 0
231 AND COALESCE(st.deleted, 0) = 0
232 )",
233 )
234 .bind(workspace_id)
235 .bind(sha256)
236 .fetch_one(&mut *tx)
237 .await?;
238 if existing {
239 tx.commit().await?;
240 return Ok(None);
241 }
242 let used: i64 = sqlx::query_scalar(
243 "SELECT COALESCE(SUM(byte_size), 0) FROM (
244 SELECT sbr.sha256, MAX(sbr.byte_size) AS byte_size
245 FROM server_blob_references sbr
246 LEFT JOIN server_task_tombstones st
247 ON st.workspace_id = sbr.workspace_id AND st.task_id = sbr.task_id
248 WHERE sbr.workspace_id = ? AND sbr.deleted = 0 AND COALESCE(st.deleted, 0) = 0
249 GROUP BY sbr.sha256
250 )",
251 )
252 .bind(workspace_id)
253 .fetch_one(&mut *tx)
254 .await?;
255 let reserved: i64 = sqlx::query_scalar(
256 "SELECT COALESCE(SUM(byte_size), 0) FROM blob_upload_reservations
257 WHERE workspace_id = ? AND sha256 != ? AND expires_at > ?",
258 )
259 .bind(workspace_id)
260 .bind(sha256)
261 .bind(timestamp(clock.now()))
262 .fetch_one(&mut *tx)
263 .await?;
264 if used.saturating_add(reserved).saturating_add(byte_size) > quota_bytes {
265 tx.rollback().await?;
266 bail!("error attachment-quota-exceeded");
267 }
268 let reservation_id = new_id();
269 let now = clock.now();
270 let expires = now + chrono::Duration::from_std(LEASE_TTL)?;
271 sqlx::query(
272 "INSERT INTO blob_upload_reservations(
273 reservation_id, workspace_id, sha256, byte_size, created_at, expires_at
274 ) VALUES (?, ?, ?, ?, ?, ?)
275 ON CONFLICT(workspace_id, sha256) DO UPDATE SET
276 reservation_id = excluded.reservation_id,
277 byte_size = excluded.byte_size,
278 created_at = excluded.created_at,
279 expires_at = excluded.expires_at",
280 )
281 .bind(&reservation_id)
282 .bind(workspace_id)
283 .bind(sha256)
284 .bind(byte_size)
285 .bind(timestamp(now))
286 .bind(timestamp(expires))
287 .execute(&mut *tx)
288 .await?;
289 tx.commit().await?;
290 Ok(Some(reservation_id))
291}
292
293pub async fn release_reservation(conn: &mut SqliteConnection, reservation_id: &str) -> Result<()> {
294 sqlx::query("DELETE FROM blob_upload_reservations WHERE reservation_id = ?")
295 .bind(reservation_id)
296 .execute(&mut *conn)
297 .await?;
298 Ok(())
299}
300
301pub async fn local_unique_bytes(conn: &mut SqliteConnection) -> Result<i64> {
302 Ok(sqlx::query_scalar(
303 "SELECT COALESCE(SUM(byte_size), 0) FROM blob_inventory WHERE available = 1",
304 )
305 .fetch_one(&mut *conn)
306 .await?)
307}
308
309pub async fn ensure_local_capacity(
310 conn: &mut SqliteConnection,
311 blob_dir: &Path,
312 sha256: &str,
313 byte_size: i64,
314 policy: LifecyclePolicy,
315 clock: &dyn Clock,
316) -> Result<Option<String>> {
317 let used = local_unique_bytes(conn).await?;
318 if used.saturating_add(byte_size) > policy.quota_bytes {
319 prune(conn, blob_dir, policy, true, clock).await?;
320 }
321 let now = clock.now();
322 let mut tx = begin_immediate(conn).await?;
323 let existing: bool = sqlx::query_scalar(
324 "SELECT EXISTS(SELECT 1 FROM blob_inventory WHERE sha256 = ? AND available = 1)",
325 )
326 .bind(sha256)
327 .fetch_one(&mut *tx)
328 .await?;
329 if existing {
330 tx.commit().await?;
331 return Ok(None);
332 }
333 let used: i64 = sqlx::query_scalar(
334 "SELECT COALESCE(SUM(byte_size), 0) FROM blob_inventory WHERE available = 1",
335 )
336 .fetch_one(&mut *tx)
337 .await?;
338 let reserved: i64 = sqlx::query_scalar(
339 "SELECT COALESCE(SUM(byte_size), 0) FROM blob_upload_reservations
340 WHERE workspace_id = '__local__' AND sha256 != ? AND expires_at > ?",
341 )
342 .bind(sha256)
343 .bind(timestamp(now))
344 .fetch_one(&mut *tx)
345 .await?;
346 if used.saturating_add(reserved).saturating_add(byte_size) > policy.quota_bytes {
347 tx.rollback().await?;
348 bail!("error attachment-quota-exceeded");
349 }
350 let reservation_id = new_id();
351 let expires = now + chrono::Duration::from_std(LEASE_TTL)?;
352 sqlx::query(
353 "INSERT INTO blob_upload_reservations(
354 reservation_id, workspace_id, sha256, byte_size, created_at, expires_at
355 ) VALUES (?, '__local__', ?, ?, ?, ?)
356 ON CONFLICT(workspace_id, sha256) DO UPDATE SET
357 reservation_id = excluded.reservation_id,
358 byte_size = excluded.byte_size,
359 created_at = excluded.created_at,
360 expires_at = excluded.expires_at",
361 )
362 .bind(&reservation_id)
363 .bind(sha256)
364 .bind(byte_size)
365 .bind(timestamp(now))
366 .bind(timestamp(expires))
367 .execute(&mut *tx)
368 .await?;
369 tx.commit().await?;
370 Ok(Some(reservation_id))
371}
372
373pub async fn reconcile_trash(conn: &mut SqliteConnection, blob_dir: &Path) -> Result<()> {
374 let trash = trash_dir(blob_dir);
375 if !trash.exists() {
376 return Ok(());
377 }
378 for entry in fs::read_dir(&trash)? {
379 let entry = entry?;
380 if !entry.file_type()?.is_file() {
381 continue;
382 }
383 let Some(sha256) = entry.file_name().to_str().map(str::to_owned) else {
384 continue;
385 };
386 if validate_sha256(&sha256).is_err() {
387 continue;
388 }
389 let available: bool = sqlx::query_scalar(
390 "SELECT COALESCE((SELECT available FROM blob_inventory WHERE sha256 = ?), 0)",
391 )
392 .bind(&sha256)
393 .fetch_one(&mut *conn)
394 .await?;
395 if available {
396 let target = object_path(blob_dir, &sha256)?;
397 if !target.exists() {
398 if let Some(parent) = target.parent() {
399 fs::create_dir_all(parent)?;
400 }
401 fs::rename(entry.path(), target)?;
402 } else {
403 fs::remove_file(entry.path())?;
404 }
405 } else {
406 fs::remove_file(entry.path())?;
407 }
408 }
409 Ok(())
410}
411
412pub async fn reconcile_staging(blob_dir: &Path) -> Result<ByteCount> {
413 let mut removed = ByteCount::default();
414 let dir = staging_dir(blob_dir);
415 if !dir.exists() {
416 return Ok(removed);
417 }
418 for entry in fs::read_dir(dir)? {
419 let entry = entry?;
420 let name = entry.file_name();
421 if !name.to_string_lossy().starts_with(".aven-stage-") {
422 continue;
423 }
424 let metadata = entry.metadata()?;
425 let stale = metadata
426 .modified()?
427 .elapsed()
428 .is_ok_and(|age| age >= LEASE_TTL);
429 if !stale {
430 continue;
431 }
432 removed.count += 1;
433 removed.bytes += metadata.len();
434 fs::remove_file(entry.path())?;
435 }
436 Ok(removed)
437}
438
439pub async fn reconcile_orphan_objects(
440 conn: &mut SqliteConnection,
441 blob_dir: &Path,
442 grace: Duration,
443 clock: &dyn Clock,
444) -> Result<ByteCount> {
445 let mut removed = ByteCount::default();
446 let dir = staging_dir(blob_dir);
447 if !dir.exists() {
448 return Ok(removed);
449 }
450 let cutoff = clock.now() - chrono::Duration::from_std(grace)?;
451 let now = timestamp(clock.now());
452 for entry in fs::read_dir(dir)? {
453 let entry = entry?;
454 if !entry.file_type()?.is_file() {
455 continue;
456 }
457 let Some(sha256) = entry.file_name().to_str().map(str::to_owned) else {
458 continue;
459 };
460 if validate_sha256(&sha256).is_err() {
461 continue;
462 }
463 let tracked: bool =
464 sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blob_inventory WHERE sha256 = ?)")
465 .bind(&sha256)
466 .fetch_one(&mut *conn)
467 .await?;
468 let metadata = entry.metadata()?;
469 if tracked
470 || DateTime::<Utc>::from(metadata.modified()?) > cutoff
471 || is_protected(conn, &sha256, &now).await?
472 {
473 continue;
474 }
475 let trash = trash_dir(blob_dir);
476 fs::create_dir_all(&trash)?;
477 let trashed = trash.join(&sha256);
478 fs::rename(entry.path(), &trashed)?;
479 fs::remove_file(trashed)?;
480 removed.count += 1;
481 removed.bytes += metadata.len();
482 }
483 Ok(removed)
484}
485
486pub async fn prune(
487 conn: &mut SqliteConnection,
488 blob_dir: &Path,
489 policy: LifecyclePolicy,
490 apply: bool,
491 clock: &dyn Clock,
492) -> Result<PruneSummary> {
493 if apply {
494 reconcile_trash(conn, blob_dir).await?;
495 reconcile_staging(blob_dir).await?;
496 }
497 reconcile_liveness(conn, clock).await?;
498 if apply {
499 reconcile_orphan_objects(conn, blob_dir, policy.grace, clock).await?;
500 }
501 let now = timestamp(clock.now());
502 let cutoff = cutoff(clock.now(), policy.grace)?;
503 let rows = sqlx::query(
504 "SELECT bi.sha256, bi.byte_size
505 FROM blob_inventory bi
506 JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256
507 WHERE bi.available = 1 AND bl.unreferenced_at IS NOT NULL
508 AND bl.unreferenced_at <= ?
509 ORDER BY bl.unreferenced_at, bi.sha256 LIMIT ?",
510 )
511 .bind(&cutoff)
512 .bind(i64::try_from(policy.maintenance_limit)?)
513 .fetch_all(&mut *conn)
514 .await?;
515 let mut summary = PruneSummary::default();
516 for row in rows {
517 let sha256: String = row.get("sha256");
518 let byte_size: i64 = row.get("byte_size");
519 if is_protected(conn, &sha256, &now).await? {
520 continue;
521 }
522 summary.eligible.count += 1;
523 summary.eligible.bytes += u64::try_from(byte_size)?;
524 if !apply {
525 continue;
526 }
527 let mut tx = begin_immediate(conn).await?;
528 let still_eligible: bool = sqlx::query_scalar(
529 "SELECT EXISTS(
530 SELECT 1 FROM blob_inventory bi
531 JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256
532 WHERE bi.sha256 = ? AND bi.available = 1
533 AND bl.unreferenced_at IS NOT NULL AND bl.unreferenced_at <= ?
534 )",
535 )
536 .bind(&sha256)
537 .bind(&cutoff)
538 .fetch_one(&mut *tx)
539 .await?;
540 if !still_eligible || is_protected(&mut tx, &sha256, &now).await? {
541 tx.rollback().await?;
542 continue;
543 }
544 let source = object_path(blob_dir, &sha256)?;
545 let trash = trash_dir(blob_dir);
546 fs::create_dir_all(&trash)?;
547 let trashed = trash.join(&sha256);
548 if source.exists() {
549 fs::rename(&source, &trashed).with_context(|| "could not move attachment to trash")?;
550 }
551 if let Err(error) = sqlx::query(
552 "UPDATE blob_inventory SET available = 0, last_verified_at = ? WHERE sha256 = ?",
553 )
554 .bind(&now)
555 .bind(&sha256)
556 .execute(&mut *tx)
557 .await
558 {
559 if trashed.exists() {
560 let _ = fs::rename(&trashed, &source);
561 }
562 return Err(error.into());
563 }
564 tx.commit().await?;
565 if trashed.exists() {
566 fs::remove_file(&trashed)?;
567 }
568 summary.pruned.count += 1;
569 summary.pruned.bytes += u64::try_from(byte_size)?;
570 }
571 prune_preview_cache(blob_dir, policy.preview_quota_bytes)?;
572 Ok(summary)
573}
574
575pub fn prune_preview_cache(blob_dir: &Path, quota: u64) -> Result<ByteCount> {
576 let root = blob_dir.join("cache").join("previews");
577 if !root.exists() {
578 return Ok(ByteCount::default());
579 }
580 let mut files = Vec::new();
581 let mut dirs = vec![root];
582 while let Some(dir) = dirs.pop() {
583 for entry in fs::read_dir(dir)? {
584 let entry = entry?;
585 if entry.file_type()?.is_dir() {
586 dirs.push(entry.path());
587 } else if entry.file_type()?.is_file() {
588 let metadata = entry.metadata()?;
589 files.push((metadata.modified()?, metadata.len(), entry.path()));
590 }
591 }
592 }
593 let mut total: u64 = files.iter().map(|(_, size, _)| *size).sum();
594 files.sort_by_key(|(modified, _, path)| (*modified, path.clone()));
595 let mut removed = ByteCount::default();
596 for (_, size, path) in files {
597 if total <= quota {
598 break;
599 }
600 fs::remove_file(path)?;
601 total -= size;
602 removed.count += 1;
603 removed.bytes += size;
604 }
605 Ok(removed)
606}
607
608pub async fn lifecycle_report(
609 conn: &mut SqliteConnection,
610 blob_dir: &Path,
611 policy: LifecyclePolicy,
612 clock: &dyn Clock,
613) -> Result<LifecycleReport> {
614 reconcile_liveness(conn, clock).await?;
615 let now = timestamp(clock.now());
616 let cutoff = cutoff(clock.now(), policy.grace)?;
617 let live_blob_references = live_blob_references_sql("bi.sha256");
618 let rows = sqlx::query(sqlx::AssertSqlSafe(format!(
619 "SELECT bi.sha256, bi.byte_size, bi.available, bl.unreferenced_at,
620 {live_blob_references} AS referenced
621 FROM blob_inventory bi LEFT JOIN blob_lifecycle bl ON bl.sha256 = bi.sha256"
622 )))
623 .fetch_all(&mut *conn)
624 .await?;
625 let mut report = LifecycleReport::default();
626 for row in rows {
627 let sha256: String = row.get("sha256");
628 let bytes = u64::try_from(row.get::<i64, _>("byte_size"))?;
629 let available = row.get::<i64, _>("available") != 0;
630 let referenced = row.get::<i64, _>("referenced") != 0;
631 let unreferenced_at: Option<String> = row.get("unreferenced_at");
632 if available {
633 report.quota.count += 1;
634 report.quota.bytes += bytes;
635 }
636 if referenced {
637 report.referenced.count += 1;
638 report.referenced.bytes += bytes;
639 } else if is_protected(conn, &sha256, &now).await? {
640 report.protected.count += 1;
641 report.protected.bytes += bytes;
642 } else if unreferenced_at
643 .as_deref()
644 .is_some_and(|at| at <= cutoff.as_str())
645 {
646 report.eligible.count += 1;
647 report.eligible.bytes += bytes;
648 } else {
649 report.grace_period.count += 1;
650 report.grace_period.bytes += bytes;
651 }
652 if referenced && unreferenced_at.is_some() || !referenced && unreferenced_at.is_none() {
653 report.inconsistencies.count += 1;
654 report.inconsistencies.bytes += bytes;
655 }
656 }
657 let (reservation_count, reservation_bytes): (i64, i64) = sqlx::query_as(
658 "SELECT COUNT(*), COALESCE(SUM(byte_size), 0)
659 FROM blob_upload_reservations WHERE expires_at > ?",
660 )
661 .bind(&now)
662 .fetch_one(&mut *conn)
663 .await?;
664 report.reservations = ByteCount {
665 count: u64::try_from(reservation_count)?,
666 bytes: u64::try_from(reservation_bytes)?,
667 };
668 for entry in fs::read_dir(staging_dir(blob_dir))
669 .into_iter()
670 .flatten()
671 .flatten()
672 {
673 let name = entry.file_name().to_string_lossy().to_string();
674 let metadata = entry.metadata()?;
675 if name.starts_with(".aven-stage-") {
676 report.staging.count += 1;
677 report.staging.bytes += metadata.len();
678 } else if validate_sha256(&name).is_ok() {
679 let tracked: bool =
680 sqlx::query_scalar("SELECT EXISTS(SELECT 1 FROM blob_inventory WHERE sha256 = ?)")
681 .bind(&name)
682 .fetch_one(&mut *conn)
683 .await?;
684 if !tracked {
685 report.staging.count += 1;
686 report.staging.bytes += metadata.len();
687 report.inconsistencies.count += 1;
688 report.inconsistencies.bytes += metadata.len();
689 }
690 }
691 }
692 for entry in fs::read_dir(trash_dir(blob_dir))
693 .into_iter()
694 .flatten()
695 .flatten()
696 {
697 if entry.file_type()?.is_file() {
698 report.trash.count += 1;
699 report.trash.bytes += entry.metadata()?.len();
700 }
701 }
702 Ok(report)
703}
704
705#[cfg(test)]
706mod tests {
707 use std::sync::{Arc, Mutex};
708
709 use chrono::DateTime;
710 use sqlx::Connection as _;
711 use sqlx::sqlite::SqliteConnectOptions;
712
713 use crate::attachments::storage::{object_path, upsert_inventory_available};
714 use crate::db::open_db;
715
716 use super::*;
717
718 #[derive(Clone)]
719 struct TestClock(Arc<Mutex<DateTime<Utc>>>);
720
721 impl TestClock {
722 fn at(value: &str) -> Self {
723 Self(Arc::new(Mutex::new(
724 DateTime::parse_from_rfc3339(value).unwrap().to_utc(),
725 )))
726 }
727
728 fn advance(&self, duration: chrono::Duration) {
729 let mut now = self.0.lock().unwrap();
730 *now += duration;
731 }
732 }
733
734 impl Clock for TestClock {
735 fn now(&self) -> DateTime<Utc> {
736 *self.0.lock().unwrap()
737 }
738 }
739
740 async fn insert_task(conn: &mut SqliteConnection, task_id: &str) {
741 sqlx::query(
742 "INSERT INTO tasks(
743 workspace_id, id, title, description, project_id, status, priority,
744 created_at, updated_at, queue_activity_at
745 ) VALUES ('0000000000000000', ?, 'task', '', 'project', 'inbox', 'none',
746 '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z', '2026-01-01T00:00:00Z')",
747 )
748 .bind(task_id)
749 .execute(conn)
750 .await
751 .unwrap();
752 }
753
754 async fn insert_attachment(
755 conn: &mut SqliteConnection,
756 attachment_id: &str,
757 task_id: &str,
758 sha256: &str,
759 deleted: bool,
760 ) {
761 sqlx::query(
762 "INSERT INTO task_attachments(
763 workspace_id, attachment_id, task_id, sha256, byte_size, media_type, width, height,
764 created_at, deleted, deleted_at
765 ) VALUES ('0000000000000000', ?, ?, ?, 4, 'image/png', 1, 1,
766 '2026-01-01T00:00:00Z', ?, ?)",
767 )
768 .bind(attachment_id)
769 .bind(task_id)
770 .bind(sha256)
771 .bind(i64::from(deleted))
772 .bind(deleted.then_some("2026-01-01T00:00:00Z"))
773 .execute(conn)
774 .await
775 .unwrap();
776 }
777
778 #[tokio::test]
779 async fn final_live_reference_starts_grace_once_and_restore_clears_it() {
780 let temp = tempfile::tempdir().unwrap();
781 let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
782 let mut conn = pool.acquire().await.unwrap();
783 let hash = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa";
784 upsert_inventory_available(&mut conn, hash, 4, "image/png")
785 .await
786 .unwrap();
787 insert_task(&mut conn, "0000000000000001").await;
788 insert_task(&mut conn, "0000000000000002").await;
789 insert_attachment(
790 &mut conn,
791 "0000000000000011",
792 "0000000000000001",
793 hash,
794 false,
795 )
796 .await;
797 insert_attachment(
798 &mut conn,
799 "0000000000000012",
800 "0000000000000002",
801 hash,
802 false,
803 )
804 .await;
805 let clock = TestClock::at("2026-07-01T00:00:00Z");
806
807 reconcile_liveness(&mut conn, &clock).await.unwrap();
808 sqlx::query("UPDATE task_attachments SET deleted = 1, deleted_at = 'x' WHERE attachment_id = '0000000000000011'")
809 .execute(&mut *conn).await.unwrap();
810 reconcile_liveness(&mut conn, &clock).await.unwrap();
811 let value: Option<String> =
812 sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
813 .bind(hash)
814 .fetch_one(&mut *conn)
815 .await
816 .unwrap();
817 assert_eq!(value, None, "one live reference keeps the hash live");
818
819 sqlx::query("UPDATE task_attachments SET deleted = 1, deleted_at = 'x' WHERE attachment_id = '0000000000000012'")
820 .execute(&mut *conn).await.unwrap();
821 reconcile_liveness(&mut conn, &clock).await.unwrap();
822 let first: String =
823 sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
824 .bind(hash)
825 .fetch_one(&mut *conn)
826 .await
827 .unwrap();
828 clock.advance(chrono::Duration::days(1));
829 reconcile_liveness(&mut conn, &clock).await.unwrap();
830 let second: String =
831 sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
832 .bind(hash)
833 .fetch_one(&mut *conn)
834 .await
835 .unwrap();
836 assert_eq!(first, second, "grace starts exactly once");
837
838 sqlx::query("UPDATE task_attachments SET deleted = 0, deleted_at = NULL WHERE attachment_id = '0000000000000012'")
839 .execute(&mut *conn).await.unwrap();
840 reconcile_liveness(&mut conn, &clock).await.unwrap();
841 let restored: Option<String> =
842 sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
843 .bind(hash)
844 .fetch_one(&mut *conn)
845 .await
846 .unwrap();
847 assert_eq!(restored, None);
848 }
849
850 #[tokio::test]
851 async fn lease_protects_expired_unreferenced_blob_until_release() {
852 let temp = tempfile::tempdir().unwrap();
853 let db_path = temp.path().join("test.sqlite");
854 let pool = open_db(&db_path).await.unwrap();
855 let mut conn = pool.acquire().await.unwrap();
856 let blob_dir = temp.path().join("blobs");
857 let hash = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb";
858 upsert_inventory_available(&mut conn, hash, 4, "image/png")
859 .await
860 .unwrap();
861 let path = object_path(&blob_dir, hash).unwrap();
862 fs::create_dir_all(path.parent().unwrap()).unwrap();
863 fs::write(&path, b"blob").unwrap();
864 let clock = TestClock::at("2026-07-10T00:00:00Z");
865 reconcile_liveness(&mut conn, &clock).await.unwrap();
866 clock.advance(chrono::Duration::days(8));
867 let lease = acquire_lease(&mut conn, hash, "backup", &clock)
868 .await
869 .unwrap();
870 let policy = LifecyclePolicy::default();
871 let blocked = prune(&mut conn, &blob_dir, policy, true, &clock)
872 .await
873 .unwrap();
874 assert_eq!(blocked.pruned.count, 0);
875 assert!(path.exists());
876
877 release_lease(&mut conn, &lease).await.unwrap();
878 let pruned = prune(&mut conn, &blob_dir, policy, true, &clock)
879 .await
880 .unwrap();
881 assert_eq!(pruned.pruned.count, 1);
882 assert!(!path.exists());
883 }
884
885 #[tokio::test]
886 async fn quota_is_unique_by_hash_and_reservations_are_workspace_scoped() {
887 let temp = tempfile::tempdir().unwrap();
888 let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
889 let mut conn = pool.acquire().await.unwrap();
890 let hash = "cccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccccc";
891 let clock = TestClock::at("2026-07-01T00:00:00Z");
892 let first = reserve_upload(&mut conn, "workspace-a", hash, 8, 8, &clock)
893 .await
894 .unwrap();
895 assert!(first.is_some());
896 let replacement = reserve_upload(&mut conn, "workspace-a", hash, 8, 8, &clock)
897 .await
898 .unwrap();
899 assert!(replacement.is_some());
900 let other = reserve_upload(&mut conn, "workspace-b", hash, 8, 8, &clock)
901 .await
902 .unwrap();
903 assert!(other.is_some());
904 let count: i64 = sqlx::query_scalar("SELECT count(*) FROM blob_upload_reservations")
905 .fetch_one(&mut *conn)
906 .await
907 .unwrap();
908 assert_eq!(count, 2);
909 }
910
911 #[tokio::test]
912 async fn concurrent_attach_wins_prune_recheck() {
913 let temp = tempfile::tempdir().unwrap();
914 let db_path = temp.path().join("test.sqlite");
915 let pool = open_db(&db_path).await.unwrap();
916 let mut conn = pool.acquire().await.unwrap();
917 let blob_dir = temp.path().join("blobs");
918 let hash = "abababababababababababababababababababababababababababababababab";
919 upsert_inventory_available(&mut conn, hash, 4, "image/png")
920 .await
921 .unwrap();
922 insert_task(&mut conn, "0000000000000003").await;
923 let path = object_path(&blob_dir, hash).unwrap();
924 fs::create_dir_all(path.parent().unwrap()).unwrap();
925 fs::write(&path, b"blob").unwrap();
926 let clock = TestClock::at("2026-07-01T00:00:00Z");
927 reconcile_liveness(&mut conn, &clock).await.unwrap();
928 clock.advance(chrono::Duration::days(8));
929 let options = SqliteConnectOptions::new()
930 .filename(&db_path)
931 .busy_timeout(Duration::from_secs(5));
932 let prune_conn = SqliteConnection::connect_with(&options).await.unwrap();
933
934 let mut tx = begin_immediate(&mut conn).await.unwrap();
935 insert_attachment(&mut tx, "0000000000000013", "0000000000000003", hash, false).await;
936 let prune_dir = blob_dir.clone();
937 let prune_clock = clock.clone();
938 let pruning = tokio::spawn(async move {
939 let mut prune_conn = prune_conn;
940 prune(
941 &mut prune_conn,
942 &prune_dir,
943 LifecyclePolicy::default(),
944 true,
945 &prune_clock,
946 )
947 .await
948 .unwrap()
949 });
950 tokio::time::sleep(Duration::from_millis(50)).await;
951 tx.commit().await.unwrap();
952
953 let summary = pruning.await.unwrap();
954 assert_eq!(summary.pruned.count, 0);
955 assert!(path.exists());
956 }
957
958 #[tokio::test]
959 async fn accepted_server_reference_keeps_blob_live() {
960 let temp = tempfile::tempdir().unwrap();
961 let pool = open_db(&temp.path().join("test.sqlite")).await.unwrap();
962 let mut conn = pool.acquire().await.unwrap();
963 let hash = "acacacacacacacacacacacacacacacacacacacacacacacacacacacacacacacac";
964 upsert_inventory_available(&mut conn, hash, 4, "image/png")
965 .await
966 .unwrap();
967 sqlx::query(
968 "INSERT INTO server_blob_references(
969 workspace_id, attachment_id, task_id, sha256, byte_size
970 ) VALUES ('workspace', '0000000000000042', '0000000000000004', ?, 4)",
971 )
972 .bind(hash)
973 .execute(&mut *conn)
974 .await
975 .unwrap();
976 let clock = TestClock::at("2026-07-01T00:00:00Z");
977
978 reconcile_liveness(&mut conn, &clock).await.unwrap();
979 let unreferenced_at: Option<String> =
980 sqlx::query_scalar("SELECT unreferenced_at FROM blob_lifecycle WHERE sha256 = ?")
981 .bind(hash)
982 .fetch_one(&mut *conn)
983 .await
984 .unwrap();
985 assert_eq!(unreferenced_at, None);
986 }
987
988 #[tokio::test]
989 async fn local_quota_boundary_is_hash_idempotent() {
990 let temp = tempfile::tempdir().unwrap();
991 let db_path = temp.path().join("test.sqlite");
992 let pool = open_db(&db_path).await.unwrap();
993 let mut conn = pool.acquire().await.unwrap();
994 let blob_dir = temp.path().join("blobs");
995 let existing = "eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee";
996 let new_hash = "ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff";
997 upsert_inventory_available(&mut conn, existing, 8, "image/png")
998 .await
999 .unwrap();
1000 let clock = TestClock::at("2026-07-01T00:00:00Z");
1001 let policy = LifecyclePolicy {
1002 quota_bytes: 8,
1003 ..LifecyclePolicy::default()
1004 };
1005
1006 ensure_local_capacity(&mut conn, &blob_dir, existing, 8, policy, &clock)
1007 .await
1008 .unwrap();
1009 let error = ensure_local_capacity(&mut conn, &blob_dir, new_hash, 1, policy, &clock)
1010 .await
1011 .unwrap_err();
1012 assert_eq!(error.to_string(), "error attachment-quota-exceeded");
1013 }
1014
1015 #[tokio::test]
1016 async fn interrupted_atomic_create_object_is_reconciled_after_grace() {
1017 let temp = tempfile::tempdir().unwrap();
1018 let db_path = temp.path().join("test.sqlite");
1019 let pool = open_db(&db_path).await.unwrap();
1020 let mut conn = pool.acquire().await.unwrap();
1021 let blob_dir = temp.path().join("blobs");
1022 let hash = "cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcd";
1023 let path = object_path(&blob_dir, hash).unwrap();
1024 fs::create_dir_all(path.parent().unwrap()).unwrap();
1025 fs::write(&path, b"orphan").unwrap();
1026 let clock = TestClock::at("2030-07-01T00:00:00Z");
1027 let policy = LifecyclePolicy {
1028 grace: Duration::ZERO,
1029 ..LifecyclePolicy::default()
1030 };
1031
1032 prune(&mut conn, &blob_dir, policy, true, &clock)
1033 .await
1034 .unwrap();
1035 assert!(!path.exists());
1036 }
1037
1038 #[tokio::test]
1039 async fn interrupted_trash_move_restores_available_object() {
1040 let temp = tempfile::tempdir().unwrap();
1041 let db_path = temp.path().join("test.sqlite");
1042 let pool = open_db(&db_path).await.unwrap();
1043 let mut conn = pool.acquire().await.unwrap();
1044 let blob_dir = temp.path().join("blobs");
1045 let hash = "dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd";
1046 upsert_inventory_available(&mut conn, hash, 4, "image/png")
1047 .await
1048 .unwrap();
1049 let source = object_path(&blob_dir, hash).unwrap();
1050 let trash = trash_dir(&blob_dir).join(hash);
1051 fs::create_dir_all(source.parent().unwrap()).unwrap();
1052 fs::create_dir_all(trash.parent().unwrap()).unwrap();
1053 fs::write(&source, b"blob").unwrap();
1054 fs::rename(&source, &trash).unwrap();
1055
1056 reconcile_trash(&mut conn, &blob_dir).await.unwrap();
1057 assert!(source.exists());
1058 assert!(!trash.exists());
1059 }
1060}