Skip to main content

max / makenotwork

server: reap abandoned multipart sessions in the orphan cleanup A multipart session that was started and never completed leaves no object to delete, only uploaded parts that S3 bills for until they are aborted. The orphan reaper only deleted objects, so its delete was a no-op against such a session and the parts leaked indefinitely. That is a cost bug, not a tidiness one, and it scales with the size of the files the multipart path exists to carry. The reaper now aborts any in-progress session for a stale pending-upload key before deleting the object. Upload ids are recovered from S3 via list_multipart_uploads_for_key rather than the database, so a session whose tracking row was lost is still cleaned up. The listing filters to an exact key match, since ListMultipartUploads matches a prefix and reaping staging/abc must not abort a live session for staging/abcdef. Aborting is best-effort: a failure logs and neither blocks the object delete nor strands the tracking row, because the bucket lifecycle rule is the backstop. InMemoryStorage now tracks sessions for real rather than stubbing them, so the reap path is observable in tests. 1887 lib + 1182 integration tests green. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-21 13:59 UTC
Commit: d85d7d092ac404357dbc5b97dffb2e604bd5ecff
Parent: 35d06a3
6 files changed, +222 insertions, -6 deletions
M server/src/scheduler/cleanup.rs +60 -2
@@ -541,6 +541,7 @@ pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
541 541 };
542 542
543 543 let mut cleaned = 0i64;
544 + let mut aborted = 0i64;
544 545 // Carry (key, bucket) pairs: the deletion is bucket-scoped so a key present
545 546 // in two buckets only clears the record for the bucket we reaped (B7).
546 547 let mut keys_to_delete: Vec<(String, String)> = Vec::with_capacity(stale.len());
@@ -558,6 +559,18 @@ pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
558 559 crate::storage::S3Bucket::Main => state.storage.s3.as_ref(),
559 560 };
560 561 if let Some(s3) = s3_client {
562 + // A multipart session that was started and never completed leaves NO
563 + // object to delete, only uploaded parts that S3 bills for until they
564 + // are aborted — so the delete below is a no-op against it and the
565 + // parts would leak forever. Abort first, keyed off S3's own list so a
566 + // session whose upload_id was never recorded is still caught.
567 + //
568 + // Best-effort: a failed abort must not block the object delete or
569 + // strand the tracking row. The bucket lifecycle rule
570 + // (AbortIncompleteMultipartUpload) is the backstop for anything
571 + // missed here.
572 + aborted += abort_orphan_multipart_sessions(s3.as_ref(), s3_key).await;
573 +
561 574 // Route through the guarded funnel: if a confirm already reclaimed
562 575 // this deterministic key, the live row owns the object and we must
563 576 // NOT delete it — just clear the stale pending_uploads record.
@@ -606,12 +619,57 @@ pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
606 619 tracing::error!(error = ?e, "failed to delete pending upload records");
607 620 }
608 621
609 - if cleaned > 0 {
610 - tracing::info!(cleaned, "cleaned up orphaned presigned uploads");
622 + if cleaned > 0 || aborted > 0 {
623 + tracing::info!(cleaned, aborted, "cleaned up orphaned presigned uploads");
611 624 }
612 625 let _ = db::scheduler_jobs::record_job_run(&state.db, "orphaned_upload_cleanup", cleaned).await;
613 626 }
614 627
628 + /// Abort every in-progress multipart session for `s3_key`, returning how many
629 + /// were aborted.
630 + ///
631 + /// Best-effort by design: the caller still deletes the object and clears the
632 + /// tracking row regardless, and the bucket's `AbortIncompleteMultipartUpload`
633 + /// lifecycle rule catches whatever fails here. Returning a count rather than a
634 + /// Result keeps a storage hiccup from stranding the rest of the reap.
635 + /// Public via a single re-export on `scheduler` (the module itself stays
636 + /// private) so integration tests can drive it against a test backend without a
637 + /// database or an `AppState`.
638 + pub async fn abort_orphan_multipart_sessions(
639 + s3: &dyn crate::storage::StorageBackend,
640 + s3_key: &str,
641 + ) -> i64 {
642 + let upload_ids = match s3.list_multipart_uploads_for_key(s3_key).await {
643 + Ok(ids) => ids,
644 + Err(e) => {
645 + tracing::warn!(
646 + s3_key = %s3_key, error = ?e,
647 + "orphan reaper: could not list multipart sessions; lifecycle rule is the backstop"
648 + );
649 + return 0;
650 + }
651 + };
652 +
653 + let key = crate::storage::S3Key::from_stored(s3_key);
654 + let mut aborted = 0i64;
655 + for upload_id in upload_ids {
656 + match s3.abort_multipart_upload(&key, &upload_id).await {
657 + Ok(()) => {
658 + aborted += 1;
659 + tracing::info!(
660 + s3_key = %s3_key, %upload_id,
661 + "orphan reaper: aborted an abandoned multipart session"
662 + );
663 + }
664 + Err(e) => tracing::warn!(
665 + s3_key = %s3_key, %upload_id, error = ?e,
666 + "orphan reaper: failed to abort multipart session; lifecycle rule is the backstop"
667 + ),
668 + }
669 + }
670 + aborted
671 + }
672 +
615 673 /// Remove stale cart items (>30 days old) and items that became unavailable.
616 674 #[tracing::instrument(skip_all, name = "scheduler::cleanup_cart_items")]
617 675 pub(super) async fn cleanup_cart_items(state: &AppState) {
M server/src/scheduler/mod.rs +4
@@ -7,6 +7,10 @@
7 7
8 8 mod announcements;
9 9 mod cleanup;
10 + /// Re-exported for integration tests: abandoned multipart sessions hold billed
11 + /// parts, so the reap path is worth exercising directly. The rest of `cleanup`
12 + /// stays private.
13 + pub use cleanup::abort_orphan_multipart_sessions;
10 14 #[doc(hidden)]
11 15 pub use cleanup::drain_pending_s3_deletions_for_test;
12 16 mod integrity;
M server/src/storage.rs +18
@@ -543,6 +543,11 @@ pub trait StorageBackend: Send + Sync {
543 543 /// pending-upload reaper calls this on sessions that were never confirmed —
544 544 /// incomplete multipart uploads bill for their parts indefinitely.
545 545 async fn abort_multipart_upload(&self, s3_key: &S3Key, upload_id: &str) -> Result<()>;
546 + /// Upload ids of the in-progress multipart sessions for exactly `s3_key`.
547 + /// The reaper recovers them from S3 rather than the database, so a session
548 + /// whose tracking row was lost is still cleaned up. Required (not defaulted):
549 + /// an empty default would silently strand billed parts.
550 + async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>>;
546 551 async fn check_connectivity(&self) -> std::result::Result<(), String>;
547 552 fn bucket(&self) -> &str;
548 553 }
@@ -1002,6 +1007,15 @@ impl S3Client {
1002 1007 .map_err(AppError::Storage)
1003 1008 }
1004 1009
1010 + /// In-progress multipart sessions for a key. See the
1011 + /// [`StorageBackend::list_multipart_uploads_for_key`] trait method.
1012 + pub async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
1013 + self.inner
1014 + .list_multipart_uploads_for_key(s3_key)
1015 + .await
1016 + .map_err(AppError::Storage)
1017 + }
1018 +
1005 1019 /// Lightweight connectivity check — issues a list with max_keys(0).
1006 1020 pub async fn check_connectivity(&self) -> std::result::Result<(), String> {
1007 1021 self.inner.check_connectivity().await
@@ -1267,6 +1281,10 @@ impl StorageBackend for S3Client {
1267 1281 self.abort_multipart_upload(s3_key, upload_id).await
1268 1282 }
1269 1283
1284 + async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
1285 + self.list_multipart_uploads_for_key(s3_key).await
1286 + }
1287 +
1270 1288 async fn check_connectivity(&self) -> std::result::Result<(), String> {
1271 1289 self.check_connectivity().await
1272 1290 }
M server/tests/harness/storage.rs +39 -4
@@ -9,6 +9,10 @@ use std::sync::Mutex;
9 9 /// Files are stored in a `HashMap<String, Vec<u8>>` behind a `Mutex`.
10 10 pub struct InMemoryStorage {
11 11 objects: Mutex<HashMap<String, Vec<u8>>>,
12 + /// Open multipart sessions: `upload_id -> s3_key`. Tracked for real (not
13 + /// stubbed) so the orphan reaper's abort path is observable in tests — an
14 + /// unaborted session is exactly the leak this models.
15 + multipart: Mutex<HashMap<String, String>>,
12 16 bucket: String,
13 17 }
14 18
@@ -17,10 +21,25 @@ impl InMemoryStorage {
17 21 pub fn new() -> Self {
18 22 InMemoryStorage {
19 23 objects: Mutex::new(HashMap::new()),
24 + multipart: Mutex::new(HashMap::new()),
20 25 bucket: "test-bucket".to_string(),
21 26 }
22 27 }
23 28
29 + /// Number of multipart sessions still open. Zero after a clean reap.
30 + pub fn open_multipart_count(&self) -> usize {
31 + self.multipart.lock().unwrap().len()
32 + }
33 +
34 + /// Open a multipart session directly, standing in for one a client started
35 + /// and abandoned.
36 + pub fn put_open_multipart(&self, upload_id: &str, s3_key: &str) {
37 + self.multipart
38 + .lock()
39 + .unwrap()
40 + .insert(upload_id.to_string(), s3_key.to_string());
41 + }
42 +
24 43 /// Pre-populate a file so that subsequent `object_exists` / `download_object`
25 44 /// calls see it. Useful for testing confirm_upload flows.
26 45 pub fn put(&self, key: &str, data: Vec<u8>) {
@@ -129,7 +148,9 @@ impl StorageBackend for InMemoryStorage {
129 148 // confirm path, exactly as they do for `presign_upload` + confirm.
130 149
131 150 async fn create_multipart_upload(&self, s3_key: &S3Key, _content_type: &str) -> Result<String> {
132 - Ok(format!("test-upload-id/{}", s3_key))
151 + let upload_id = format!("test-upload-id/{}", s3_key);
152 + self.put_open_multipart(&upload_id, s3_key.as_str());
153 + Ok(upload_id)
133 154 }
134 155
135 156 async fn presign_upload_part(&self, s3_key: &S3Key, upload_id: &str, part_number: i32, _expiry_secs: Option<u64>, _max_bytes: Option<i64>) -> Result<String> {
@@ -144,20 +165,34 @@ impl StorageBackend for InMemoryStorage {
144 165 Ok(format!("http://test-storage/{s3_key}?uploadId={upload_id}&partNumber={part_number}"))
145 166 }
146 167
147 - async fn complete_multipart_upload(&self, _s3_key: &S3Key, _upload_id: &str, parts: &[(i32, String)]) -> Result<()> {
168 + async fn complete_multipart_upload(&self, _s3_key: &S3Key, upload_id: &str, parts: &[(i32, String)]) -> Result<()> {
148 169 if parts.is_empty() {
149 170 return Err(AppError::Storage(
150 171 "cannot complete a multipart upload with no parts".to_string(),
151 172 ));
152 173 }
174 + // Completing closes the session, so it no longer holds billed parts.
175 + self.multipart.lock().unwrap().remove(upload_id);
153 176 Ok(())
154 177 }
155 178
156 - async fn abort_multipart_upload(&self, _s3_key: &S3Key, _upload_id: &str) -> Result<()> {
157 - // Idempotent: nothing is staged, so aborting an unknown session is fine.
179 + async fn abort_multipart_upload(&self, _s3_key: &S3Key, upload_id: &str) -> Result<()> {
180 + // Idempotent: aborting an unknown session is fine.
181 + self.multipart.lock().unwrap().remove(upload_id);
158 182 Ok(())
159 183 }
160 184
185 + async fn list_multipart_uploads_for_key(&self, s3_key: &str) -> Result<Vec<String>> {
186 + Ok(self
187 + .multipart
188 + .lock()
189 + .unwrap()
190 + .iter()
191 + .filter(|(_, key)| key.as_str() == s3_key)
192 + .map(|(id, _)| id.clone())
193 + .collect())
194 + }
195 +
161 196 async fn check_connectivity(&self) -> std::result::Result<(), String> {
162 197 Ok(())
163 198 }
M server/tests/workflows/storage.rs +42
@@ -1835,3 +1835,45 @@ async fn multipart_complete_then_confirm_commits_the_upload() {
1835 1835 .unwrap();
1836 1836 assert_eq!(storage_used, size, "confirm must charge the real object size");
1837 1837 }
1838 +
1839 + #[tokio::test]
1840 + async fn orphan_reaper_aborts_abandoned_multipart_sessions() {
1841 + // A multipart session that was started and never completed leaves NO object
1842 + // to delete, only uploaded parts that S3 bills for until they are aborted.
1843 + // The reaper's object delete is a no-op against it, so without this abort
1844 + // the parts leak indefinitely — a cost bug, not just a tidiness one.
1845 + use makenotwork::scheduler::abort_orphan_multipart_sessions;
1846 +
1847 + let mem = crate::harness::storage::InMemoryStorage::new();
1848 + let target = "staging/abandoned-uuid/movie.mp4";
1849 +
1850 + // Two abandoned sessions on the reaped key, plus one on a different key
1851 + // that must survive: ListMultipartUploads matches a PREFIX, so a careless
1852 + // implementation reaping `staging/abc` would also kill `staging/abcdef`.
1853 + mem.put_open_multipart("upload-a", target);
1854 + mem.put_open_multipart("upload-b", target);
1855 + mem.put_open_multipart("upload-c", "staging/abandoned-uuid/movie.mp4.other");
1856 + assert_eq!(mem.open_multipart_count(), 3);
1857 +
1858 + let aborted = abort_orphan_multipart_sessions(&mem, target).await;
1859 +
1860 + assert_eq!(aborted, 2, "both sessions on the reaped key must be aborted");
1861 + assert_eq!(
1862 + mem.open_multipart_count(),
1863 + 1,
1864 + "a session on a different key must survive the prefix-adjacent reap"
1865 + );
1866 + }
1867 +
1868 + #[tokio::test]
1869 + async fn orphan_reaper_abort_is_a_noop_when_there_is_no_session() {
1870 + // The common case: an ordinary single-PUT orphan has no multipart session,
1871 + // and the reaper must sail past it without erroring or miscounting.
1872 + use makenotwork::scheduler::abort_orphan_multipart_sessions;
1873 +
1874 + let mem = crate::harness::storage::InMemoryStorage::new();
1875 + mem.put("staging/plain/song.mp3", b"bytes".to_vec());
1876 +
1877 + let aborted = abort_orphan_multipart_sessions(&mem, "staging/plain/song.mp3").await;
1878 + assert_eq!(aborted, 0);
1879 + }
M shared/s3-storage/src/lib.rs +59
@@ -905,6 +905,65 @@ impl S3Client {
905 905 .map_err(|e| format!("S3 abort multipart upload for {key} failed: {e}"))
906 906 }
907 907
908 + /// List the in-progress multipart uploads for exactly `key`, returning their
909 + /// upload ids.
910 + ///
911 + /// The orphan reaper needs this: a session that was started but never
912 + /// completed has **no object to delete**, only uploaded parts that S3 bills
913 + /// for until they are aborted, so a plain `delete` is a no-op against it. The
914 + /// upload id is not recorded durably anywhere, so it is recovered from S3,
915 + /// which also catches sessions whose tracking row was lost entirely.
916 + ///
917 + /// `ListMultipartUploads` matches a *prefix*, so results are filtered to an
918 + /// exact key match — otherwise reaping `staging/abc` would also abort a live
919 + /// session for `staging/abcdef`.
920 + pub async fn list_multipart_uploads_for_key(&self, key: &str) -> Result<Vec<String>, String> {
921 + let mut ids = Vec::new();
922 + let mut key_marker: Option<String> = None;
923 + let mut upload_id_marker: Option<String> = None;
924 +
925 + loop {
926 + let mut req = self
927 + .client
928 + .list_multipart_uploads()
929 + .bucket(&self.bucket)
930 + .prefix(key);
931 + if let Some(ref k) = key_marker {
932 + req = req.key_marker(k);
933 + }
934 + if let Some(ref u) = upload_id_marker {
935 + req = req.upload_id_marker(u);
936 + }
937 +
938 + let resp = req
939 + .send()
940 + .await
941 + .map_err(|e| format!("S3 list_multipart_uploads for {key} failed: {e}"))?;
942 +
943 + for upload in resp.uploads() {
944 + // Exact-key filter: the request matched on prefix.
945 + if upload.key() == Some(key)
946 + && let Some(id) = upload.upload_id()
947 + {
948 + ids.push(id.to_string());
949 + }
950 + }
951 +
952 + if resp.is_truncated().unwrap_or(false) {
953 + key_marker = resp.next_key_marker().map(str::to_string);
954 + upload_id_marker = resp.next_upload_id_marker().map(str::to_string);
955 + // Defensive: a truncated response with no markers would loop forever.
956 + if key_marker.is_none() && upload_id_marker.is_none() {
957 + break;
958 + }
959 + } else {
960 + break;
961 + }
962 + }
963 +
964 + Ok(ids)
965 + }
966 +
908 967 /// Server-side copy an object using multipart `UploadPartCopy`, for sources
909 968 /// larger than the 5 GiB single-part [`Self::copy_object`]/`CopyObject`
910 969 /// limit. No bytes transit the caller — S3 copies each byte range internally.