Skip to main content

max / makenotwork

Reach the orphaned-upload reaper from a test The reaper was `pub(super)` and only the scheduler tick called it, so no test could enter it. That included the branch handing a failed S3 delete to the durable deletion queue before clearing the tracking row, which is the fix for a leak that already shipped once (Run #2 Storage SERIOUS): clearing the row on a transient failure dropped the only record of the object. Code written to close a known leak, unobserved ever since. Add a `_for_test` entry point alongside the existing one, keep the assembled state on the harness to call it with, and cover three branches: a clean reap, a transient delete failure that must be queued for retry before its row goes, and a failing multipart abort that must not block the delete it precedes. Checked by reintroducing the leak: routing `GuardedDelete::Failed` back into `keys_to_delete` fails the middle test on the queued-for-retry assertion, which is the regression itself. Phase 1 of wiki `testing-posture`.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-05 14:13 UTC
Signed with PGP, not checked
Commit: 90c96f89e9294888594455b708353308a3b5ed57
Parent: c435c5c
4 files changed, +148 insertions, -1 deletion
M server/src/scheduler/cleanup.rs +14
@@ -887,6 +887,20 @@
887 887
888 888 /// Test-only synchronous drain of the pending-S3-deletion queue (`main` bucket).
889 889 ///
890 + /// Run [`cleanup_orphaned_uploads`] once, synchronously.
891 + ///
892 + /// The reaper is `pub(super)` and the scheduler drives it on a tick, so no
893 + /// integration test could reach it, including the branch that hands a failed S3
894 + /// delete to the durable deletion queue instead of dropping the tracking row.
895 + /// That branch is the fix for a leak that shipped once already (Run #2 Storage
896 + /// SERIOUS), which is exactly the kind of code that should not be reachable only
897 + /// in production. Exposed via `TestHarness::run_orphan_upload_reaper`.
898 + #[doc(hidden)]
899 + #[tracing::instrument(skip_all, name = "scheduler::cleanup_orphaned_uploads_for_test")]
900 + pub async fn cleanup_orphaned_uploads_for_test(state: &AppState) {
901 + cleanup_orphaned_uploads(state).await;
902 + }
903 +
890 904 /// The S3 delete a confirm/delete handler triggers is asynchronous: handlers
891 905 /// only [`enqueue_s3_orphan`](crate::routes::storage::enqueue_s3_orphan), and
892 906 /// the scheduler's [`retry_pending_s3_deletions`] performs the actual delete
M server/src/scheduler/mod.rs +1 -1
@@ -12,7 +12,7 @@
12 12 /// stays private.
13 13 pub use cleanup::abort_orphan_multipart_sessions;
14 14 #[doc(hidden)]
15 - pub use cleanup::drain_pending_s3_deletions_for_test;
15 + pub use cleanup::{cleanup_orphaned_uploads_for_test, drain_pending_s3_deletions_for_test};
16 16 mod integrity;
17 17 mod mt_threads;
18 18 mod synckit_warnings;
M server/tests/harness/mod.rs +15
@@ -119,6 +119,9 @@
119 119 /// Pieces needed to drain the scan worker synchronously from tests
120 120 /// (`drain_scan_jobs`). `None` when the harness wasn't built with a scanner.
121 121 scan_deps: Option<ScanDeps>,
122 + /// The assembled state, kept so tests can drive scheduler jobs that take it.
123 + /// `build_app` borrows rather than consumes it, so this costs a cheap clone.
124 + state: makenotwork::AppState,
122 125 _test_db: TestDb,
123 126 }
124 127
@@ -526,10 +529,22 @@
526 529 mock_email: mock_email_ref,
527 530 mock_stripe,
528 531 scan_deps,
532 + state,
529 533 _test_db: test_db,
530 534 }
531 535 }
532 536
537 + /// Run the orphaned-upload reaper once, synchronously.
538 + ///
539 + /// The scheduler drives this on a tick in production. Tests that want the
540 + /// reaper's failure branches (a transient S3 delete handed off to the
541 + /// durable queue, a best-effort multipart abort) need it to run at a known
542 + /// point instead, the same reason `drain_s3_deletions` exists.
543 + #[allow(dead_code)]
544 + pub(crate) async fn run_orphan_upload_reaper(&self) {
545 + makenotwork::scheduler::cleanup_orphaned_uploads_for_test(&self.state).await;
546 + }
547 +
533 548 /// Sign up a new user via POST /join. Returns the user's ID.
534 549 pub(crate) async fn signup(&mut self, username: &str, email: &str, password: &str) -> UserId {
535 550 // Fetch a page first to establish session + CSRF
M server/tests/workflows/failure_paths.rs +118
@@ -587,3 +587,121 @@
587 587 "a retired job must never be claimed again"
588 588 );
589 589 }
590 +
591 + // The orphaned-upload reaper
592 +
593 + /// Insert a pending upload that is already old enough for the reaper, with the
594 + /// object present in storage. Returns the key.
595 + async fn stale_pending_upload(h: &TestHarness, user_id: db::UserId, key: &str) -> String {
596 + h.storage.as_ref().unwrap().put(key, b"orphan".to_vec());
597 + sqlx::query(
598 + "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at)
599 + VALUES ($1, $2, 'main', NOW() - INTERVAL '48 hours')",
600 + )
601 + .bind(user_id)
602 + .bind(key)
603 + .execute(&h.db)
604 + .await
605 + .unwrap();
606 + key.to_string()
607 + }
608 +
609 + async fn pending_upload_rows(h: &TestHarness, key: &str) -> i64 {
610 + sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE s3_key = $1")
611 + .bind(key)
612 + .fetch_one(&h.db)
613 + .await
614 + .unwrap()
615 + }
616 +
617 + /// The happy path, asserted here so the failure path below is a contrast rather
618 + /// than the only thing observed: a reaped orphan is deleted, its tracking row is
619 + /// cleared, and nothing is handed to the durable queue.
620 + #[tokio::test]
621 + async fn the_reaper_deletes_an_orphan_and_clears_its_row() {
622 + let mut h = TestHarness::with_storage().await;
623 + let user_id = h.signup("reap1", "reap1@test.com", "pass1234").await;
624 + let key = stale_pending_upload(&h, user_id, "staging/reaped.bin").await;
625 + let storage = h.storage.clone().unwrap();
626 +
627 + h.run_orphan_upload_reaper().await;
628 +
629 + assert!(
630 + !storage.object_exists(&key).await.unwrap(),
631 + "the orphan object is deleted"
632 + );
633 + assert_eq!(
634 + pending_upload_rows(&h, &key).await,
635 + 0,
636 + "tracking row cleared"
637 + );
638 + assert_eq!(
639 + queued_deletions(&h, &key).await,
640 + 0,
641 + "a successful delete must not also enqueue, that would double-handle the key"
642 + );
643 + }
644 +
645 + /// A transient S3 failure must hand the key to the durable deletion queue
646 + /// BEFORE the tracking row is cleared. Clearing the row on a transient failure
647 + /// dropped the only record of the object and leaked it permanently (Run #2
648 + /// Storage SERIOUS). The fix has been in the tree unobserved since; this is the
649 + /// test that enters it.
650 + #[tokio::test]
651 + async fn a_transient_delete_failure_hands_the_orphan_to_the_durable_queue() {
652 + let mut h = TestHarness::with_storage().await;
653 + let user_id = h.signup("reap2", "reap2@test.com", "pass1234").await;
654 + let key = stale_pending_upload(&h, user_id, "staging/handed-off.bin").await;
655 + let storage = h.storage.clone().unwrap();
656 +
657 + storage
658 + .faults()
659 + .fail_always("delete_object", storage_unavailable);
660 + h.run_orphan_upload_reaper().await;
661 +
662 + assert!(
663 + storage.object_exists(&key).await.unwrap(),
664 + "the delete failed, so the object is still there"
665 + );
666 + assert_eq!(
667 + queued_deletions(&h, &key).await,
668 + 1,
669 + "the key must be queued for retry; without this the object leaks"
670 + );
671 + assert_eq!(
672 + pending_upload_rows(&h, &key).await,
673 + 0,
674 + "the tracking row is cleared only because the durable queue now owns the key"
675 + );
676 +
677 + // The handoff is worth nothing if the queue cannot then finish the job.
678 + storage.faults().clear("delete_object");
679 + assert_eq!(h.drain_s3_deletions().await, 1, "the retry completes it");
680 + assert!(!storage.object_exists(&key).await.unwrap(), "object gone");
681 + }
682 +
683 + /// Aborting orphaned multipart sessions is documented best-effort: it must not
684 + /// block the object delete. A failing abort that stranded the delete would leave
685 + /// the orphan in place every tick forever, and the tracking row with it.
686 + #[tokio::test]
687 + async fn a_failed_multipart_abort_does_not_block_the_orphan_delete() {
688 + let mut h = TestHarness::with_storage().await;
689 + let user_id = h.signup("reap3", "reap3@test.com", "pass1234").await;
690 + let key = stale_pending_upload(&h, user_id, "staging/abort-fails.bin").await;
691 + let storage = h.storage.clone().unwrap();
692 +
693 + storage
694 + .faults()
695 + .fail_always("list_multipart_uploads_for_key", storage_unavailable);
696 + h.run_orphan_upload_reaper().await;
697 +
698 + assert!(
699 + !storage.object_exists(&key).await.unwrap(),
700 + "a failed abort is best-effort and must not stop the delete"
701 + );
702 + assert_eq!(
703 + pending_upload_rows(&h, &key).await,
704 + 0,
705 + "and the tracking row is still cleared"
706 + );
707 + }