teamcraft/services/
retention.rs

1//! Data retention: which vault blobs the database still refers to, the
2//! removal of blobs nothing refers to (`teamcraft vault gc`, and the blobs of
3//! expired staged imports and upload sessions) and the trimming of the
4//! preview cache, and the removal of superseded versions of `latest` files
5//! ([`purge_superseded_versions`]). The audit log retention is
6//! `audit::purge_older_than`.
7//!
8//! **Referenced content is never removed.** A blob is removed only when no
9//! column of [`BLOB_COLUMNS`] names it *and* it was not written or stored
10//! again within a grace period: the vault renews a blob's modification time
11//! whenever the same content is stored again (`vault.rs`), and every writer
12//! stores the blob before it inserts the row that refers to it, so content on
13//! its way into the database is always younger than the grace period. The
14//! time is checked again right before the removal. A store of identical
15//! content that overlaps the removal itself re-checks that the blob exists
16//! after renewing it and puts its own copy in place otherwise; the remaining
17//! window is a few instructions wide and only open for blobs that were idle
18//! for the whole grace period.
19
20use std::{
21    collections::HashSet,
22    path::{Path, PathBuf},
23    time::{Duration, SystemTime},
24};
25
26use sqlx::{PgConnection, PgPool};
27
28use serde_json::json;
29use uuid::Uuid;
30
31use crate::{
32    domain::FileRetention,
33    error::{AppError, AppResult},
34    preview,
35    services::{audit, revisions},
36    vault::Vault,
37};
38
39/// Every column that may name a vault blob by its SHA-256 – the files and
40/// their versions, change request attachments, staged STEP imports and
41/// upload sessions (the stored content, and the hashes the client announced
42/// or started from). `shape_signatures.sha256` is derived data and
43/// `branding_logos.sha256` the hash of bytes kept in the row; neither keeps a
44/// blob alive. A test checks this list against the migrations.
45pub const BLOB_COLUMNS: &[(&str, &str)] = &[
46    ("change_request_attachments", "sha256"),
47    ("file_versions", "sha256"),
48    ("files", "sha256"),
49    ("staged_imports", "sha256"),
50    ("upload_sessions", "base_sha256"),
51    ("upload_sessions", "sha256"),
52    ("upload_sessions", "stored_sha256"),
53];
54
55/// Hash columns that are not references to vault content.
56pub const NOT_BLOB_COLUMNS: &[(&str, &str)] = &[
57    // the logo bytes are stored in the row itself (`data`)
58    ("branding_logos", "sha256"),
59    ("shape_signatures", "sha256"),
60];
61
62/// Grace period of the housekeeping: blobs of expired staged imports and
63/// upload sessions written or stored again within the last hour stay (and
64/// the staged import with them, until the next run).
65pub const IN_FLIGHT_GRACE: Duration = Duration::from_secs(3600);
66
67/// Default grace period of `teamcraft vault gc`.
68pub const DEFAULT_GC_GRACE_HOURS: u64 = 24;
69
70/// Hashes asked about per query.
71const BATCH: usize = 1000;
72
73/// Temporary files of the preview cache older than this are leftovers.
74const PREVIEW_TEMP_AGE: Duration = Duration::from_secs(3600);
75
76/// The hashes of `hashes` that a column of [`BLOB_COLUMNS`] names.
77pub async fn referenced(conn: &mut PgConnection, hashes: &[String]) -> AppResult<HashSet<String>> {
78    let mut found = HashSet::new();
79    for chunk in hashes.chunks(BATCH) {
80        let union = BLOB_COLUMNS
81            .iter()
82            .map(|(table, column)| {
83                format!("SELECT {column} AS sha256 FROM {table} WHERE {column} = ANY($1)")
84            })
85            .collect::<Vec<_>>()
86            .join(" UNION ");
87        let rows: Vec<String> = sqlx::query_scalar(&union)
88            .bind(chunk)
89            .fetch_all(&mut *conn)
90            .await?;
91        found.extend(rows);
92    }
93    Ok(found)
94}
95
96/// What [`release_blob`] did.
97#[derive(Clone, Copy, Debug, PartialEq, Eq)]
98pub enum Release {
99    /// Nothing referred to the blob any more; it is gone.
100    Removed,
101    /// The database still refers to it: kept.
102    Referenced,
103    /// Written or stored again within the grace period: kept for now.
104    Recent,
105    /// It was not in the vault (any more).
106    Missing,
107}
108
109/// Removes one blob if nothing in the database refers to it – as seen by
110/// `conn`, so a caller can delete the last referring row in the same
111/// transaction first – and it is older than `grace`.
112pub async fn release_blob(
113    conn: &mut PgConnection,
114    vault: &Vault,
115    sha256: &str,
116    grace: Duration,
117) -> AppResult<Release> {
118    let sha256 = sha256.to_string();
119    if !referenced(conn, std::slice::from_ref(&sha256))
120        .await?
121        .is_empty()
122    {
123        return Ok(Release::Referenced);
124    }
125    match vault.object_age(&sha256).await? {
126        None => Ok(Release::Missing),
127        Some(age) if age < grace => Ok(Release::Recent),
128        Some(_) if vault.remove_object_older_than(&sha256, grace).await? => Ok(Release::Removed),
129        // renewed (or removed) in the meantime
130        Some(_) => Ok(Release::Recent),
131    }
132}
133
134/// An unreferenced blob found by [`collect_garbage`].
135#[derive(Clone, Debug)]
136pub struct Orphan {
137    pub sha256: String,
138    pub size: u64,
139    /// Whether it was removed (only with `apply`; a blob stored again
140    /// meanwhile stays).
141    pub removed: bool,
142}
143
144/// Outcome of [`collect_garbage`].
145#[derive(Clone, Debug, Default)]
146pub struct GcReport {
147    /// Blobs in the vault.
148    pub scanned: u64,
149    /// Blobs the database refers to.
150    pub referenced: u64,
151    /// Unreferenced blobs within the grace period (kept).
152    pub recent: u64,
153    /// Unreferenced blobs older than the grace period.
154    pub orphans: Vec<Orphan>,
155}
156
157impl GcReport {
158    pub fn orphan_bytes(&self) -> u64 {
159        self.orphans.iter().map(|o| o.size).sum()
160    }
161
162    pub fn removed(&self) -> (u64, u64) {
163        self.orphans
164            .iter()
165            .filter(|o| o.removed)
166            .fold((0, 0), |(n, bytes), o| (n + 1, bytes + o.size))
167    }
168}
169
170/// `teamcraft vault gc`: walks the vault one `objects/ab/` directory at a
171/// time and finds the blobs no column of [`BLOB_COLUMNS`] names and that were
172/// not written or stored again within `grace`. Only with `apply` are they
173/// removed (each after a last check of its time).
174pub async fn collect_garbage(
175    pool: &PgPool,
176    vault: &Vault,
177    grace: Duration,
178    apply: bool,
179) -> AppResult<GcReport> {
180    let mut report = GcReport::default();
181    let mut conn = pool.acquire().await?;
182    let now = SystemTime::now();
183    for prefix in vault.object_prefixes().await? {
184        let objects = vault.objects_in(&prefix).await?;
185        report.scanned += objects.len() as u64;
186        let hashes: Vec<String> = objects.iter().map(|o| o.sha256.clone()).collect();
187        let known = referenced(&mut conn, &hashes).await?;
188        for object in objects {
189            if known.contains(&object.sha256) {
190                report.referenced += 1;
191                continue;
192            }
193            let age = now.duration_since(object.modified).unwrap_or_default();
194            if age < grace {
195                report.recent += 1;
196                continue;
197            }
198            let removed = apply
199                && vault
200                    .remove_object_older_than(&object.sha256, grace)
201                    .await?;
202            report.orphans.push(Orphan {
203                sha256: object.sha256,
204                size: object.size,
205                removed,
206            });
207        }
208    }
209    Ok(report)
210}
211
212/// Outcome of [`purge_previews`].
213#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)]
214pub struct PreviewPurge {
215    /// Cache files removed.
216    pub removed: u64,
217    pub removed_bytes: u64,
218    /// Size of the cache files kept.
219    pub kept_bytes: u64,
220}
221
222struct CacheFile {
223    path: PathBuf,
224    sha256: String,
225    size: u64,
226    used: SystemTime,
227}
228
229/// Trims the preview cache (`<vault>/previews`): removes the previews and
230/// thumbnails of blobs the database no longer refers to, files of older
231/// cache versions and temporary files older than an hour; then, with
232/// `max_bytes` > 0, the least recently used files until the cache fits.
233/// Files with other names are left alone. Everything removed is made again
234/// when it is needed.
235pub async fn purge_previews(pool: &PgPool, dir: &Path, max_bytes: u64) -> AppResult<PreviewPurge> {
236    let mut purge = PreviewPurge::default();
237    let mut kept = Vec::new();
238    let mut conn = pool.acquire().await?;
239    let mut tops = match tokio::fs::read_dir(dir).await {
240        Ok(tops) => tops,
241        Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(purge),
242        Err(e) => return Err(e.into()),
243    };
244    let now = SystemTime::now();
245    while let Some(top) = tops.next_entry().await? {
246        if !top.file_type().await?.is_dir() {
247            continue;
248        }
249        let mut current = Vec::new();
250        let mut files = tokio::fs::read_dir(top.path()).await?;
251        while let Some(file) = files.next_entry().await? {
252            let meta = file.metadata().await?;
253            if !meta.is_file() {
254                continue;
255            }
256            let name = file.file_name();
257            let Some(name) = name.to_str() else {
258                continue;
259            };
260            let modified = meta.modified().unwrap_or(now);
261            match preview::cache_entry(name) {
262                preview::CacheEntry::Current(sha256) => current.push(CacheFile {
263                    path: file.path(),
264                    sha256: sha256.to_string(),
265                    size: meta.len(),
266                    // a read updates the access time (at least daily with
267                    // `relatime`); without one the write time counts
268                    used: meta.accessed().map_or(modified, |a| a.max(modified)),
269                }),
270                preview::CacheEntry::Stale => {
271                    remove_cache_file(&file.path(), meta.len(), &mut purge).await;
272                }
273                preview::CacheEntry::Temporary
274                    if now.duration_since(modified).unwrap_or_default() > PREVIEW_TEMP_AGE =>
275                {
276                    remove_cache_file(&file.path(), meta.len(), &mut purge).await;
277                }
278                preview::CacheEntry::Temporary | preview::CacheEntry::Unknown => {}
279            }
280        }
281        let hashes: Vec<String> = current.iter().map(|f| f.sha256.clone()).collect();
282        let known = referenced(&mut conn, &hashes).await?;
283        for f in current {
284            if known.contains(&f.sha256) {
285                kept.push(f);
286            } else {
287                remove_cache_file(&f.path, f.size, &mut purge).await;
288            }
289        }
290    }
291    let mut total: u64 = kept.iter().map(|f| f.size).sum();
292    if max_bytes > 0 && total > max_bytes {
293        kept.sort_by_key(|f| f.used);
294        for f in &kept {
295            if total <= max_bytes {
296                break;
297            }
298            if remove_cache_file(&f.path, f.size, &mut purge).await {
299                total -= f.size;
300            }
301        }
302    }
303    purge.kept_bytes = total;
304    Ok(purge)
305}
306
307async fn remove_cache_file(path: &Path, size: u64, purge: &mut PreviewPurge) -> bool {
308    match tokio::fs::remove_file(path).await {
309        Ok(()) => {
310            purge.removed += 1;
311            purge.removed_bytes += size;
312            true
313        }
314        // gone already, or open on Windows: the next run tries again
315        Err(_) => false,
316    }
317}
318
319/// Default grace period before a superseded version of a `latest` file is
320/// removed (`TEAMCRAFT_RETENTION_LATEST_GRACE_DAYS`).
321pub const DEFAULT_LATEST_GRACE_DAYS: u32 = 7;
322
323/// Longest grace period accepted (ten years).
324pub const MAX_LATEST_GRACE_DAYS: u32 = 3650;
325
326/// The superseded versions of one file that [`purge_superseded_versions`]
327/// removed (or, in a dry run, would remove).
328#[derive(Clone, Debug)]
329pub struct PurgedFile {
330    pub item_number: String,
331    pub revision: String,
332    pub filename: String,
333    pub file_id: Uuid,
334    pub versions: Vec<i32>,
335    pub bytes: u64,
336}
337
338/// Outcome of [`purge_superseded_versions`].
339#[derive(Clone, Debug, Default)]
340pub struct VersionPurge {
341    pub files: Vec<PurgedFile>,
342    /// Vault blobs removed because nothing referred to them any more (always
343    /// 0 in a dry run; content other files still use stays).
344    pub blobs_removed: u64,
345}
346
347impl VersionPurge {
348    pub fn versions(&self) -> u64 {
349        self.files.iter().map(|f| f.versions.len() as u64).sum()
350    }
351
352    pub fn bytes(&self) -> u64 {
353        self.files.iter().map(|f| f.bytes).sum()
354    }
355}
356
357/// Removes the old versions of files whose retention is `latest`: a version
358/// goes when a newer version replaced it at least `grace_days` ago (the time
359/// of the next version's upload counts, so a rollback or a mistaken upload
360/// can still be undone within the grace period). The blobs nothing refers to
361/// afterwards are removed from the vault. Without `apply` it only reports.
362///
363/// # Invariants
364///
365/// * **Permission:** none – the server's housekeeping and
366///   `teamcraft vault purge-versions` (an operator with the configuration),
367///   no actor ("system").
368/// * **Locks:** per file one transaction with the revision row and the file
369///   row (`FOR UPDATE`), the same order as an upload; the conditions are
370///   checked again under the locks.
371/// * **Data:** only files with `retention = 'latest'` whose revision is
372///   *in work*: nothing of an in-review, released or obsolete revision, which
373///   is therefore never touched; the current version is never removed; a
374///   grace period of 0 days is refused (400).
375/// * **Audit:** one `file.purge_versions` entry per file, in the transaction
376///   of the removal, naming the versions and hashes.
377/// * **Atomic:** per file; the vault blobs are removed after the commit
378///   (checked against every reference and the in-flight grace period).
379pub async fn purge_superseded_versions(
380    pool: &PgPool,
381    vault: &Vault,
382    grace_days: u32,
383    apply: bool,
384) -> AppResult<VersionPurge> {
385    if grace_days == 0 || grace_days > MAX_LATEST_GRACE_DAYS {
386        return Err(AppError::BadRequest(format!(
387            "The grace period must be between 1 and {MAX_LATEST_GRACE_DAYS} days."
388        )));
389    }
390    let days = i32::try_from(grace_days).unwrap_or(i32::MAX);
391    let candidates: Vec<Uuid> = sqlx::query_scalar(
392        "SELECT f.id FROM files f JOIN revisions r ON r.id = f.revision_id
393         WHERE f.retention = 'latest' AND r.state = 'in_work' AND f.current_version > 1
394         ORDER BY f.id",
395    )
396    .fetch_all(pool)
397    .await?;
398    let mut out = VersionPurge::default();
399    let mut shas = Vec::new();
400    for file_id in candidates {
401        let mut tx = pool.begin().await?;
402        let revision_id: Option<Uuid> =
403            sqlx::query_scalar("SELECT revision_id FROM files WHERE id = $1")
404                .bind(file_id)
405                .fetch_optional(&mut *tx)
406                .await?;
407        let Some(revision_id) = revision_id else {
408            continue;
409        };
410        let rev = match revisions::lock(&mut tx, revision_id).await {
411            Ok(rev) => rev,
412            Err(AppError::NotFound(_)) => continue,
413            Err(e) => return Err(e),
414        };
415        let current: Option<(i32, String, String)> = sqlx::query_as(
416            "SELECT current_version, retention, filename FROM files WHERE id = $1 FOR UPDATE",
417        )
418        .bind(file_id)
419        .fetch_optional(&mut *tx)
420        .await?;
421        let Some((current, retention, filename)) = current else {
422            continue;
423        };
424        if retention != FileRetention::Latest.as_str() || rev.state != "in_work" {
425            continue;
426        }
427        let old: Vec<(i32, String, i64)> = sqlx::query_as(
428            "WITH s AS (
429                 SELECT version, sha256, size_bytes,
430                        lead(uploaded_at) OVER (ORDER BY version) AS superseded_at
431                 FROM file_versions WHERE file_id = $1
432             )
433             SELECT version, sha256, size_bytes FROM s
434             WHERE version < $2 AND superseded_at < now() - make_interval(days => $3)
435             ORDER BY version",
436        )
437        .bind(file_id)
438        .bind(current)
439        .bind(days)
440        .fetch_all(&mut *tx)
441        .await?;
442        if old.is_empty() {
443            continue;
444        }
445        let versions: Vec<i32> = old.iter().map(|(v, _, _)| *v).collect();
446        let bytes: u64 = old
447            .iter()
448            .map(|(_, _, size)| u64::try_from(*size).unwrap_or_default())
449            .sum();
450        if apply {
451            sqlx::query("DELETE FROM file_versions WHERE file_id = $1 AND version = ANY($2)")
452                .bind(file_id)
453                .bind(&versions)
454                .execute(&mut *tx)
455                .await?;
456            audit::record(
457                &mut tx,
458                None,
459                "file.purge_versions",
460                "revision",
461                Some(rev.id),
462                format!(
463                    "Removed {} superseded version(s) of {filename} from {} (retention: latest)",
464                    versions.len(),
465                    revisions::label(&rev)
466                ),
467                json!({
468                    "item_id": rev.item_id,
469                    "file_id": file_id,
470                    "filename": filename,
471                    "versions": versions,
472                    "sha256": old.iter().map(|(_, sha, _)| sha.as_str()).collect::<Vec<_>>(),
473                    "bytes": bytes,
474                    "grace_days": grace_days,
475                    "kept_version": current,
476                }),
477            )
478            .await?;
479            shas.extend(old.iter().map(|(_, sha, _)| sha.clone()));
480        }
481        tx.commit().await?;
482        out.files.push(PurgedFile {
483            item_number: rev.item_number.clone(),
484            revision: rev.code.clone(),
485            filename,
486            file_id,
487            versions,
488            bytes,
489        });
490    }
491    shas.sort();
492    shas.dedup();
493    for sha256 in shas {
494        let mut conn = pool.acquire().await?;
495        if release_blob(&mut conn, vault, &sha256, IN_FLIGHT_GRACE).await? == Release::Removed {
496            out.blobs_removed += 1;
497        }
498    }
499    Ok(out)
500}
501
502#[cfg(test)]
503mod tests {
504    use super::*;
505    use std::collections::BTreeSet;
506
507    /// `(table, column)` of every TEXT column of the migrations whose name
508    /// ends in `sha256`.
509    fn hash_columns_of_the_migrations() -> BTreeSet<(String, String)> {
510        let dir = Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
511        let table =
512            regex::Regex::new(r"(?i)^\s*(?:CREATE|ALTER)\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?(\w+)")
513                .unwrap();
514        let column = regex::Regex::new(r"(?i)\b(\w*sha256)\s+TEXT\b").unwrap();
515        let mut found = BTreeSet::new();
516        let mut paths: Vec<_> = std::fs::read_dir(dir)
517            .unwrap()
518            .map(|e| e.unwrap().path())
519            .filter(|p| p.extension().is_some_and(|e| e == "sql"))
520            .collect();
521        paths.sort();
522        for path in paths {
523            let text = std::fs::read_to_string(&path).unwrap();
524            let mut current = String::new();
525            for line in text.lines() {
526                // without the comment
527                let line = line.split("--").next().unwrap_or_default();
528                if let Some(c) = table.captures(line) {
529                    current = c[1].to_lowercase();
530                }
531                // also `ALTER TABLE t ADD COLUMN x_sha256 TEXT` on one line
532                for c in column.captures_iter(line) {
533                    found.insert((current.clone(), c[1].to_lowercase()));
534                }
535            }
536        }
537        found
538    }
539
540    #[test]
541    fn every_hash_column_is_classified() {
542        let found = hash_columns_of_the_migrations();
543        assert!(
544            found.contains(&("files".into(), "sha256".into())),
545            "the scanner sees the columns: {found:?}"
546        );
547        let listed: BTreeSet<(String, String)> = BLOB_COLUMNS
548            .iter()
549            .chain(NOT_BLOB_COLUMNS)
550            .map(|(t, c)| (t.to_string(), c.to_string()))
551            .collect();
552        let unlisted: Vec<_> = found.difference(&listed).collect();
553        assert!(
554            unlisted.is_empty(),
555            "a migration adds a hash column: list it in BLOB_COLUMNS (it names vault content, \
556             which must never be removed while it is referenced) or in NOT_BLOB_COLUMNS: \
557             {unlisted:?}"
558        );
559        let stale: Vec<_> = listed.difference(&found).collect();
560        assert!(stale.is_empty(), "no such column: {stale:?}");
561    }
562}
teamcraft/services/retention.rs

teamcraft/services/
retention.rs
