headless_lms_server/programs/
exercise_spec_upload_reaper.rs

1//! Removes files uploaded through the exercise-service upload route that never reached a saved
2//! spec.
3//!
4//! Nothing else can reclaim them: the host stores specs as opaque blobs, so a stored file's only
5//! reference may sit inside content it cannot parse. What makes a file safe here is a declaration
6//! — `exercise_task_spec_files` for a live spec, `page_history_spec_files` for every snapshot a
7//! restore could bring back. Because history is kept, this reaper mostly collects uploads a
8//! teacher abandoned before saving, not files dropped from a spec that was once saved.
9//!
10//! The binding row is soft-deleted rather than removed, leaving an audit trail of what was
11//! reclaimed.
12
13use std::{env, path::Path};
14
15use crate::config::{FileStoreRuntimeConfig, program_config::ProgramConfig};
16use crate::{setup_file_store, setup_tracing};
17use dotenvy::dotenv;
18use futures::{StreamExt, stream};
19use headless_lms_models::{self as models, error::TryToOptional};
20use headless_lms_utils::file_store::FileStore;
21use sqlx::{PgConnection, PgPool};
22
23const MAX_CONCURRENT_REAPS: usize = 8;
24
25pub async fn main() -> anyhow::Result<()> {
26    // TODO: Audit that the environment access only happens in single-threaded code.
27    unsafe { env::set_var("RUST_LOG", "info,actix_web=info,sqlx=warn") };
28    dotenv().ok();
29    setup_tracing()?;
30    let database_url = ProgramConfig::database_url_with_default();
31    let base_url = ProgramConfig::required("BASE_URL")?;
32    let file_store = setup_file_store(&FileStoreRuntimeConfig::try_from_env()?, &base_url).await;
33    let db_pool = PgPool::connect(&database_url).await?;
34    reap(&db_pool, file_store.as_ref()).await
35}
36
37async fn reap(pool: &PgPool, file_store: &dyn FileStore) -> anyhow::Result<()> {
38    let mut conn = pool.acquire().await?;
39    let reapable = models::exercise_spec_uploads::get_reapable(&mut conn).await?;
40    drop(conn);
41    info!("Reaping {} abandoned spec uploads.", reapable.len());
42
43    let mut reaped = 0;
44    let mut skipped = 0;
45    let mut failed = 0;
46    let mut results = stream::iter(reapable)
47        .map(|upload| async move {
48            let file_upload_id = upload.file_upload_id;
49            let result = match pool.acquire().await {
50                Ok(mut conn) => reap_one(&mut conn, file_store, &upload).await,
51                Err(err) => Err(err.into()),
52            };
53            (file_upload_id, result)
54        })
55        .buffer_unordered(MAX_CONCURRENT_REAPS);
56    while let Some((file_upload_id, result)) = results.next().await {
57        match result {
58            Ok(true) => reaped += 1,
59            Ok(false) => skipped += 1,
60            Err(err) => {
61                failed += 1;
62                error!("Failed to reap spec upload {}: {:#?}", file_upload_id, err);
63            }
64        }
65    }
66    info!(
67        "Abandoned spec uploads reaped. Succeeded: {reaped}, skipped: {skipped}, failed: {failed}."
68    );
69    // The CronJob's exit status is the only signal anyone watches, so a run where every delete
70    // failed must not look green.
71    if failed > 0 {
72        anyhow::bail!(
73            "Failed to reap {failed} of {} spec uploads.",
74            reaped + failed
75        );
76    }
77    Ok(())
78}
79
80/// Retires the record, removes the object, and only then soft-deletes the `file_uploads` row.
81/// `Ok(false)` means a save came to declare the file after `get_reapable` listed it, so it is no
82/// longer reapable.
83///
84/// Deleting the `file_uploads` row last is what makes a failed object delete recoverable:
85/// `get_reapable` still sees the row and retries it on a later run, instead of orphaning the
86/// object forever.
87async fn reap_one(
88    conn: &mut PgConnection,
89    file_store: &dyn FileStore,
90    upload: &models::exercise_spec_uploads::ReapableUpload,
91) -> anyhow::Result<bool> {
92    if !models::exercise_spec_uploads::mark_reaped(conn, upload.id).await? {
93        return Ok(false);
94    }
95    file_store.delete(Path::new(&upload.path)).await?;
96    // `optional` tolerates a file a previous interrupted run already soft-deleted, without
97    // swallowing real database errors.
98    models::file_uploads::delete_and_fetch_path(conn, upload.file_upload_id)
99        .await
100        .optional()?;
101    Ok(true)
102}
headless_lms_server/programs/exercise_spec_upload_reaper.rs

headless_lms_server/programs/
exercise_spec_upload_reaper.rs
