headless_lms_server/programs/
exercise_answer_upload_reaper.rs

1//! Removes files uploaded to be named in an exercise answer that no submission was ever made
2//! from.
3//!
4//! How long an upload is spared depends on its origin, since a native client uploads immediately
5//! before submitting while an iframe student may hold an upload for the length of an exam. The
6//! binding row is soft-deleted rather than removed so that a submit naming a reaped file can
7//! still answer `upload_expired` instead of the misleading `unknown_upload`.
8
9use std::{env, path::Path};
10
11use crate::config::{FileStoreRuntimeConfig, program_config::ProgramConfig};
12use crate::{setup_file_store, setup_tracing};
13use dotenvy::dotenv;
14use futures::{StreamExt, stream};
15use headless_lms_models::{self as models, error::TryToOptional};
16use headless_lms_utils::file_store::FileStore;
17use sqlx::{PgConnection, PgPool};
18
19const MAX_CONCURRENT_REAPS: usize = 8;
20
21pub async fn main() -> anyhow::Result<()> {
22    // TODO: Audit that the environment access only happens in single-threaded code.
23    unsafe { env::set_var("RUST_LOG", "info,actix_web=info,sqlx=warn") };
24    dotenv().ok();
25    setup_tracing()?;
26    let database_url = ProgramConfig::database_url_with_default();
27    let base_url = ProgramConfig::required("BASE_URL")?;
28    let file_store = setup_file_store(&FileStoreRuntimeConfig::try_from_env()?, &base_url).await;
29    let db_pool = PgPool::connect(&database_url).await?;
30    reap(&db_pool, file_store.as_ref()).await
31}
32
33async fn reap(pool: &PgPool, file_store: &dyn FileStore) -> anyhow::Result<()> {
34    let mut conn = pool.acquire().await?;
35    let reapable = models::exercise_answer_uploads::get_reapable(&mut conn).await?;
36    drop(conn);
37    info!("Reaping {} orphaned answer uploads.", reapable.len());
38
39    let mut reaped = 0;
40    let mut skipped = 0;
41    let mut failed = 0;
42    let mut results = stream::iter(reapable)
43        .map(|upload| async move {
44            let file_upload_id = upload.file_upload_id;
45            let result = match pool.acquire().await {
46                Ok(mut conn) => reap_one(&mut conn, file_store, &upload).await,
47                Err(err) => Err(err.into()),
48            };
49            (file_upload_id, result)
50        })
51        .buffer_unordered(MAX_CONCURRENT_REAPS);
52    while let Some((file_upload_id, result)) = results.next().await {
53        match result {
54            Ok(true) => reaped += 1,
55            Ok(false) => skipped += 1,
56            Err(err) => {
57                failed += 1;
58                error!(
59                    "Failed to reap answer upload {}: {:#?}",
60                    file_upload_id, err
61                );
62            }
63        }
64    }
65    info!(
66        "Orphaned answer uploads reaped. Succeeded: {reaped}, skipped: {skipped}, failed: {failed}."
67    );
68    // The CronJob's exit status is the only signal anyone watches, so a run where every delete
69    // failed must not look green.
70    if failed > 0 {
71        anyhow::bail!(
72            "Failed to reap {failed} of {} answer uploads.",
73            reaped + failed
74        );
75    }
76    Ok(())
77}
78
79/// Retires the binding, removes the object, and only then soft-deletes the `file_uploads` row.
80/// `Ok(false)` means a submission came to reference the upload after `get_reapable` listed it, so
81/// it is no longer reapable.
82///
83/// The order matters in both directions. Retiring the binding first means a submit naming this
84/// upload answers `upload_expired` rather than succeeding and handing the exercise service a URL
85/// that 404s. Deleting the `file_uploads` row last means `get_reapable` still sees the row after a
86/// failed object delete and retries it on a later run, instead of orphaning the object forever.
87async fn reap_one(
88    conn: &mut PgConnection,
89    file_store: &dyn FileStore,
90    upload: &models::exercise_answer_uploads::ReapableUpload,
91) -> anyhow::Result<bool> {
92    if !models::exercise_answer_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}
103
104#[cfg(test)]
105mod tests {
106    use super::*;
107    use crate::test_helper::*;
108    use chrono::Duration;
109    use headless_lms_base::error::backend_error::BackendError;
110    use headless_lms_utils::prelude::{UtilError, UtilErrorType, UtilResult};
111    use std::sync::{LazyLock, Mutex};
112
113    /// `reap` is global: it takes every eligible row in the shared test database, so two of
114    /// these tests in flight at once reap each other's committed fixtures. Held for the whole test,
115    /// fixtures included. Tokio's mutex rather than `std`'s, so one failing test does not poison
116    /// the lock and fail the rest.
117    static REAPER_TESTS: LazyLock<tokio::sync::Mutex<()>> =
118        LazyLock::new(|| tokio::sync::Mutex::new(()));
119
120    /// Records what the reaper asked it to delete. The real stores are irrelevant here: the
121    /// property under test is which paths the reaper touches.
122    #[derive(Default)]
123    struct RecordingFileStore {
124        deleted: Mutex<Vec<String>>,
125    }
126
127    #[async_trait::async_trait(?Send)]
128    impl FileStore for RecordingFileStore {
129        async fn upload(&self, _path: &Path, _contents: Vec<u8>, _mime: &str) -> UtilResult<()> {
130            unimplemented!("not reached by the reaper")
131        }
132
133        async fn upload_stream(
134            &self,
135            _path: &Path,
136            _contents: headless_lms_utils::file_store::GenericPayload,
137            _mime: &str,
138        ) -> UtilResult<()> {
139            unimplemented!("not reached by the reaper")
140        }
141
142        async fn download(&self, _path: &Path) -> UtilResult<Vec<u8>> {
143            unimplemented!("not reached by the reaper")
144        }
145
146        async fn download_stream(
147            &self,
148            _path: &Path,
149        ) -> UtilResult<Box<dyn futures::Stream<Item = std::io::Result<bytes::Bytes>>>> {
150            unimplemented!("not reached by the reaper")
151        }
152
153        async fn get_direct_download_url(&self, _path: &Path) -> UtilResult<String> {
154            unimplemented!("not reached by the reaper")
155        }
156
157        async fn delete(&self, path: &Path) -> UtilResult<()> {
158            self.deleted
159                .lock()
160                .expect("lock")
161                .push(path.to_string_lossy().to_string());
162            Ok(())
163        }
164
165        fn get_cache_files_folder_path(&self) -> UtilResult<&Path> {
166            unimplemented!("not reached by the reaper")
167        }
168    }
169
170    /// How many times the reaper touched `path`. Scoped this way because the tests commit their
171    /// fixtures, so every run also sees the rows of whatever sibling test is running beside it.
172    fn deletions_of(recorded: &Mutex<Vec<String>>, path: &str) -> usize {
173        recorded
174            .lock()
175            .expect("lock")
176            .iter()
177            .filter(|recorded_path| recorded_path.as_str() == path)
178            .count()
179    }
180
181    /// Covers the whole per-row sequence: the object goes, the file row is soft-deleted, and the
182    /// binding survives soft-deleted so submit can still answer `upload_expired`.
183    ///
184    /// Committed, since `reap` now reaps each row over its own pool connection rather than the
185    /// connection fixtures were inserted on.
186    #[actix_web::test]
187    async fn reaps_an_orphaned_upload_and_leaves_a_row_behind() {
188        let _serialized = REAPER_TESTS.lock().await;
189        insert_data!(:tx, user: user, :org, :course, instance: _instance, :course_module, :chapter, :page, :exercise, :slide, task: _task);
190        let path = "exercise-services-client/orphan";
191        let file_id = models::file_uploads::insert(
192            tx.as_mut(),
193            "orphan.tar.zst",
194            path,
195            "application/octet-stream",
196            Some(user),
197            None,
198        )
199        .await
200        .expect("file upload");
201        models::exercise_answer_uploads::insert_many(
202            tx.as_mut(),
203            exercise,
204            user,
205            &[file_id],
206            models::exercise_answer_uploads::AnswerUploadOrigin::NativeClient,
207        )
208        .await
209        .expect("binding");
210        backdate(tx.as_mut(), file_id, Duration::hours(2)).await;
211        tx.commit().await;
212
213        let pool = PgPool::connect(&test_database_url())
214            .await
215            .expect("test pool");
216        let file_store = RecordingFileStore::default();
217        reap(&pool, &file_store).await.expect("reap");
218
219        assert_eq!(deletions_of(&file_store.deleted, path), 1);
220        let mut check_conn = Conn::init().await;
221        let mut check_tx = check_conn.begin().await;
222        assert!(
223            models::file_uploads::get_many(check_tx.as_mut(), &[file_id])
224                .await
225                .expect("file lookup")
226                .is_empty()
227        );
228        let recorded = models::exercise_answer_uploads::get_for_exercise_and_user(
229            check_tx.as_mut(),
230            exercise,
231            user,
232            &[file_id],
233        )
234        .await
235        .expect("binding lookup");
236        assert_eq!(recorded.len(), 1);
237        assert!(recorded[0].deleted);
238        check_tx.rollback().await;
239    }
240
241    /// Committed, for the same reason as above.
242    #[actix_web::test]
243    async fn spares_an_upload_inside_the_retention_window() {
244        let _serialized = REAPER_TESTS.lock().await;
245        insert_data!(:tx, user: user, :org, :course, instance: _instance, :course_module, :chapter, :page, :exercise, :slide, task: _task);
246        let file_id = models::file_uploads::insert(
247            tx.as_mut(),
248            "fresh.tar.zst",
249            "exercise-services-client/fresh",
250            "application/octet-stream",
251            Some(user),
252            None,
253        )
254        .await
255        .expect("file upload");
256        models::exercise_answer_uploads::insert_many(
257            tx.as_mut(),
258            exercise,
259            user,
260            &[file_id],
261            models::exercise_answer_uploads::AnswerUploadOrigin::NativeClient,
262        )
263        .await
264        .expect("binding");
265        tx.commit().await;
266
267        let pool = PgPool::connect(&test_database_url())
268            .await
269            .expect("test pool");
270        let file_store = RecordingFileStore::default();
271        reap(&pool, &file_store).await.expect("reap");
272
273        assert_eq!(
274            deletions_of(&file_store.deleted, "exercise-services-client/fresh"),
275            0
276        );
277        let mut check_conn = Conn::init().await;
278        let mut check_tx = check_conn.begin().await;
279        assert_eq!(
280            models::file_uploads::get_many(check_tx.as_mut(), &[file_id])
281                .await
282                .expect("file lookup")
283                .len(),
284            1
285        );
286        check_tx.rollback().await;
287    }
288
289    /// The origin split, end to end: two hours is past a native client's window but nowhere near an
290    /// iframe student's, who may still be holding the file mid-exam.
291    ///
292    /// Committed, for the same reason as the tests above.
293    #[actix_web::test]
294    async fn spares_an_iframe_upload_a_native_client_upload_would_lose() {
295        let _serialized = REAPER_TESTS.lock().await;
296        insert_data!(:tx, user: user, :org, :course, instance: _instance, :course_module, :chapter, :page, :exercise, :slide, task: _task);
297        let path = "exercise-services-client/mid-exam";
298        let file_id = models::file_uploads::insert(
299            tx.as_mut(),
300            "mid-exam.pdf",
301            path,
302            "application/pdf",
303            Some(user),
304            None,
305        )
306        .await
307        .expect("file upload");
308        models::exercise_answer_uploads::insert_many(
309            tx.as_mut(),
310            exercise,
311            user,
312            &[file_id],
313            models::exercise_answer_uploads::AnswerUploadOrigin::Iframe,
314        )
315        .await
316        .expect("binding");
317        backdate(tx.as_mut(), file_id, Duration::hours(2)).await;
318        tx.commit().await;
319
320        let pool = PgPool::connect(&test_database_url())
321            .await
322            .expect("test pool");
323        let file_store = RecordingFileStore::default();
324        reap(&pool, &file_store).await.expect("reap");
325
326        assert_eq!(deletions_of(&file_store.deleted, path), 0);
327        let mut check_conn = Conn::init().await;
328        let mut check_tx = check_conn.begin().await;
329        assert_eq!(
330            models::file_uploads::get_many(check_tx.as_mut(), &[file_id])
331                .await
332                .expect("file lookup")
333                .len(),
334            1
335        );
336        check_tx.rollback().await;
337    }
338
339    /// Committed, for the same reason as the tests above.
340    #[actix_web::test]
341    async fn reaps_an_iframe_upload_past_seven_days() {
342        let _serialized = REAPER_TESTS.lock().await;
343        insert_data!(:tx, user: user, :org, :course, instance: _instance, :course_module, :chapter, :page, :exercise, :slide, task: _task);
344        let path = "exercise-services-client/abandoned";
345        let file_id = models::file_uploads::insert(
346            tx.as_mut(),
347            "abandoned.pdf",
348            path,
349            "application/pdf",
350            Some(user),
351            None,
352        )
353        .await
354        .expect("file upload");
355        models::exercise_answer_uploads::insert_many(
356            tx.as_mut(),
357            exercise,
358            user,
359            &[file_id],
360            models::exercise_answer_uploads::AnswerUploadOrigin::Iframe,
361        )
362        .await
363        .expect("binding");
364        backdate(tx.as_mut(), file_id, Duration::days(8)).await;
365        tx.commit().await;
366
367        let pool = PgPool::connect(&test_database_url())
368            .await
369            .expect("test pool");
370        let file_store = RecordingFileStore::default();
371        reap(&pool, &file_store).await.expect("reap");
372
373        assert_eq!(deletions_of(&file_store.deleted, path), 1);
374        let mut check_conn = Conn::init().await;
375        let mut check_tx = check_conn.begin().await;
376        assert!(
377            models::file_uploads::get_many(check_tx.as_mut(), &[file_id])
378                .await
379                .expect("file lookup")
380                .is_empty()
381        );
382        check_tx.rollback().await;
383    }
384
385    /// Fails every delete, standing in for a transient object-store error.
386    #[derive(Default)]
387    struct FailingFileStore {
388        attempts: Mutex<Vec<String>>,
389    }
390
391    #[async_trait::async_trait(?Send)]
392    impl FileStore for FailingFileStore {
393        async fn upload(&self, _path: &Path, _contents: Vec<u8>, _mime: &str) -> UtilResult<()> {
394            unimplemented!("not reached by the reaper")
395        }
396
397        async fn upload_stream(
398            &self,
399            _path: &Path,
400            _contents: headless_lms_utils::file_store::GenericPayload,
401            _mime: &str,
402        ) -> UtilResult<()> {
403            unimplemented!("not reached by the reaper")
404        }
405
406        async fn download(&self, _path: &Path) -> UtilResult<Vec<u8>> {
407            unimplemented!("not reached by the reaper")
408        }
409
410        async fn download_stream(
411            &self,
412            _path: &Path,
413        ) -> UtilResult<Box<dyn futures::Stream<Item = std::io::Result<bytes::Bytes>>>> {
414            unimplemented!("not reached by the reaper")
415        }
416
417        async fn get_direct_download_url(&self, _path: &Path) -> UtilResult<String> {
418            unimplemented!("not reached by the reaper")
419        }
420
421        async fn delete(&self, path: &Path) -> UtilResult<()> {
422            self.attempts
423                .lock()
424                .expect("lock")
425                .push(path.to_string_lossy().to_string());
426            Err(UtilError::new(
427                UtilErrorType::Other,
428                "simulated object store failure".to_string(),
429                None,
430            ))
431        }
432
433        fn get_cache_files_folder_path(&self) -> UtilResult<&Path> {
434            unimplemented!("not reached by the reaper")
435        }
436    }
437
438    /// A failed object delete must surface in the exit status and be retried later. Before this
439    /// was fixed the run was green and the object was orphaned forever, because the binding's own
440    /// `deleted_at` excluded the row from every later listing.
441    ///
442    /// Committed, for the same reason as the tests above.
443    #[actix_web::test]
444    async fn a_failed_object_delete_fails_the_run_and_is_retried() {
445        let _serialized = REAPER_TESTS.lock().await;
446        insert_data!(:tx, user: user, :org, :course, instance: _instance, :course_module, :chapter, :page, :exercise, :slide, task: _task);
447        let path = "exercise-services-client/transient";
448        let file_id = models::file_uploads::insert(
449            tx.as_mut(),
450            "transient.tar.zst",
451            path,
452            "application/octet-stream",
453            Some(user),
454            None,
455        )
456        .await
457        .expect("file upload");
458        models::exercise_answer_uploads::insert_many(
459            tx.as_mut(),
460            exercise,
461            user,
462            &[file_id],
463            models::exercise_answer_uploads::AnswerUploadOrigin::NativeClient,
464        )
465        .await
466        .expect("binding");
467        backdate(tx.as_mut(), file_id, Duration::hours(2)).await;
468        tx.commit().await;
469
470        let pool = PgPool::connect(&test_database_url())
471            .await
472            .expect("test pool");
473        let failing = FailingFileStore::default();
474        reap(&pool, &failing)
475            .await
476            .expect_err("a run where every delete failed must not exit successfully");
477        assert_eq!(deletions_of(&failing.attempts, path), 1);
478        let mut check_conn = Conn::init().await;
479        let mut check_tx = check_conn.begin().await;
480        assert_eq!(
481            models::file_uploads::get_many(check_tx.as_mut(), &[file_id])
482                .await
483                .expect("file lookup")
484                .len(),
485            1,
486            "the file row must survive, since it is what makes the retry possible"
487        );
488        check_tx.rollback().await;
489
490        let retry = RecordingFileStore::default();
491        reap(&pool, &retry).await.expect("retry");
492        assert_eq!(deletions_of(&retry.deleted, path), 1);
493        let mut check_conn = Conn::init().await;
494        let mut check_tx = check_conn.begin().await;
495        assert!(
496            models::file_uploads::get_many(check_tx.as_mut(), &[file_id])
497                .await
498                .expect("file lookup")
499                .is_empty()
500        );
501        check_tx.rollback().await;
502    }
503
504    /// The reap-vs-submit race with two real connections, which is the part
505    /// `models::exercise_answer_uploads`' own tests cannot reach: they run inside one
506    /// uncommitted transaction, so a second connection can never see their fixtures.
507    ///
508    /// What is under test is not just the outcome but the mechanism — that a reaper running
509    /// concurrently with a submit *blocks* on the row lock `lock_for_exercise_and_user` takes,
510    /// instead of racing past it, and that when it unblocks its `NOT EXISTS` re-check observes the
511    /// association the submit committed. The second half is a Postgres detail worth pinning: the
512    /// blocked `UPDATE` re-evaluates its qual, subquery included, against the committed row, so it
513    /// declines rather than destroying the files of a submission that already returned 200.
514    #[actix_web::test]
515    async fn a_concurrent_reaper_blocks_on_the_submit_lock_and_then_declines_to_reap() {
516        let _serialized = REAPER_TESTS.lock().await;
517        // Committed so the reaper's connection can see them. Deliberately not backdated: an
518        // upload inside the retention window is invisible to `get_reapable`, so what this test
519        // leaves in the database cannot perturb the unfiltered `reap()` calls above.
520        insert_data!(:tx, user: user, :org, course: course, instance: _instance, :course_module, :chapter, :page, :exercise, slide: slide, task: task);
521        let file_id = models::file_uploads::insert(
522            tx.as_mut(),
523            "raced.tar.zst",
524            "exercise-services-client/raced",
525            "application/octet-stream",
526            Some(user),
527            None,
528        )
529        .await
530        .expect("file upload");
531        models::exercise_answer_uploads::insert_many(
532            tx.as_mut(),
533            exercise,
534            user,
535            &[file_id],
536            models::exercise_answer_uploads::AnswerUploadOrigin::NativeClient,
537        )
538        .await
539        .expect("binding");
540        let binding_id = binding_id_of(tx.as_mut(), file_id).await;
541        tx.commit().await;
542
543        // The submit side: validate under the row lock, inside the transaction that will record
544        // the association.
545        let mut submit_conn = Conn::init().await;
546        let mut submit_tx = submit_conn.begin().await;
547        let locked = models::exercise_answer_uploads::lock_for_exercise_and_user(
548            submit_tx.as_mut(),
549            exercise,
550            user,
551            &[file_id],
552        )
553        .await
554        .expect("locked lookup");
555        assert_eq!(locked.len(), 1);
556        assert!(!locked[0].deleted);
557
558        // The reaper, on its own connection, tries to retire the very row the submit holds.
559        let mut reaper_conn = Conn::init().await;
560        let mut reaper_tx = reaper_conn.begin().await;
561        // Scoped so the pinned future releases its borrow of `reaper_tx` before the rollback.
562        let reaped = {
563            let mut reap = std::pin::pin!(models::exercise_answer_uploads::mark_reaped(
564                reaper_tx.as_mut(),
565                binding_id
566            ));
567            assert!(
568                tokio::time::timeout(std::time::Duration::from_millis(500), &mut reap)
569                    .await
570                    .is_err(),
571                "the reaper must block on the row lock the submit holds, not decide without it"
572            );
573
574            let slide_submission = models::exercise_slide_submissions::insert_exercise_slide_submission(
575            submit_tx.as_mut(),
576            models::exercise_slide_submissions::NewExerciseSlideSubmission {
577                exercise_slide_id: slide,
578                course_id: Some(course),
579                exam_id: None,
580                user_id: user,
581                exercise_id: exercise,
582                user_points_update_strategy:
583                    models::exercise_task_gradings::UserPointsUpdateStrategy::CanAddPointsAndCanRemovePoints,
584            },
585        )
586        .await
587        .expect("slide submission");
588            let task_submission = models::exercise_task_submissions::insert(
589                submit_tx.as_mut(),
590                models::PKeyPolicy::Generate,
591                slide_submission.id,
592                slide,
593                task,
594                &models::library::grading::SubmittedAnswer::Json {
595                    data: serde_json::json!({ "opaque": "plugin owned" }),
596                },
597            )
598            .await
599            .expect("task submission");
600            models::exercise_task_submission_files::insert_many(
601                submit_tx.as_mut(),
602                task_submission,
603                &[file_id],
604            )
605            .await
606            .expect("submission files");
607            submit_tx.commit().await;
608
609            tokio::time::timeout(std::time::Duration::from_secs(10), &mut reap)
610                .await
611                .expect("the reaper must unblock once the submit commits")
612                .expect("mark_reaped")
613        };
614        assert!(
615            !reaped,
616            "the reaper must decline an upload the submit referenced while it waited"
617        );
618        reaper_tx.rollback().await;
619
620        let mut check_conn = Conn::init().await;
621        let mut check_tx = check_conn.begin().await;
622        let recorded = models::exercise_answer_uploads::get_for_exercise_and_user(
623            check_tx.as_mut(),
624            exercise,
625            user,
626            &[file_id],
627        )
628        .await
629        .expect("binding lookup");
630        assert_eq!(
631            recorded,
632            vec![models::exercise_answer_uploads::AnswerUpload {
633                file_upload_id: file_id,
634                deleted: false
635            }],
636            "the upload must stay usable, so download_submission can still serve it"
637        );
638        check_tx.rollback().await;
639    }
640
641    async fn binding_id_of(conn: &mut PgConnection, file_upload_id: uuid::Uuid) -> uuid::Uuid {
642        models::exercise_answer_uploads::get_id_by_file_upload_id(conn, file_upload_id)
643            .await
644            .expect("binding id")
645    }
646
647    async fn backdate(conn: &mut PgConnection, file_upload_id: uuid::Uuid, age: Duration) {
648        models::exercise_answer_uploads::backdate(conn, file_upload_id, age)
649            .await
650            .expect("backdate");
651    }
652}
headless_lms_server/programs/exercise_answer_upload_reaper.rs

headless_lms_server/programs/
exercise_answer_upload_reaper.rs
