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

headless_lms_server/programs/
exercise_answer_upload_reaper.rs
