Skip to main content

max / makenotwork

Type the S3 bucket noun and harden multipart/delete (ultra-fuzz Run 11 Storage) Drive the Storage axis to A+: - s3-storage multipart now reads each part once into an owned Bytes and clones a refcount per retry instead of re-copying buf[..n].to_vec() inside the loop — removes the ~2x part_size resident. complete_multipart_upload gains the same 3-attempt backoff as the part uploads (losing completion orphans every already- uploaded part). delete_objects reports malformed keys as failures instead of silently dropping them, so a caller can't believe an undeletable key was reaped. - New storage::S3Bucket enum is the single source of truth for the "main"/"synckit" spellings. enqueue_s3_orphan requires an S3Bucket, not a bare string, so a future synckit caller can't mis-tag an object as main (where the worker would delete it against the wrong client and leak it). The deletion worker dispatches via S3Bucket::from_db_str instead of stringly matches. - The prefix-delete liveness guard is now symmetric across buckets: a synckit {app_id}/ or ota/{app_id}/ wipe is refused when the sync app still exists, like the main {user_id}/ guard. item-cover confirm pins the full {user}/{item}/cover/ prefix (was {user}/{item}/), symmetric with the version-key path. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-30 17:57 UTC
Commit: fb4ac6e146fc4c9107e09a56053f1d99fc8189e4
Parent: 1f0c86a
11 files changed, +214 insertions, -92 deletions
M server/src/routes/api/content_insertions.rs +3 -3
@@ -137,7 +137,7 @@ pub(super) async fn confirm_insertion(
137 137
138 138 // Enforce static per-type size limit
139 139 if file_size_bytes as u64 > FileType::Insertion.max_size() {
140 - crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, "insertion_upload_rejected").await;
140 + crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "insertion_upload_rejected").await;
141 141 return Err(AppError::BadRequest(format!(
142 142 "File exceeds maximum size of {} MB",
143 143 FileType::Insertion.max_size() / (1024 * 1024)
@@ -158,7 +158,7 @@ pub(super) async fn confirm_insertion(
158 158 let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Insertion, file_size_bytes).await {
159 159 Ok(max) => max,
160 160 Err(e) => {
161 - crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, "insertion_upload_rejected").await;
161 + crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "insertion_upload_rejected").await;
162 162 return Err(e);
163 163 }
164 164 };
@@ -166,7 +166,7 @@ pub(super) async fn confirm_insertion(
166 166 // Atomically increment storage BEFORE writing the DB record.
167 167 // Avoids orphaned unbilled file references.
168 168 if let Err(e) = db::creator_tiers::try_increment_storage(&state.db, user.id, file_size_bytes, max_storage).await {
169 - crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, "insertion_upload_rejected").await;
169 + crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "insertion_upload_rejected").await;
170 170 return Err(e);
171 171 }
172 172
M server/src/routes/api/internal/uploads.rs +4 -4
@@ -143,7 +143,7 @@ pub(super) async fn confirm_upload(
143 143 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
144 144 })?;
145 145 if file_size_bytes as u64 > file_type.max_size() {
146 - crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, "cli_upload_rejected").await;
146 + crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "cli_upload_rejected").await;
147 147 return Err(AppError::BadRequest(format!(
148 148 "File exceeds maximum size of {} MB",
149 149 file_type.max_size() / (1024 * 1024)
@@ -161,7 +161,7 @@ pub(super) async fn confirm_upload(
161 161 {
162 162 Ok(max) => max,
163 163 Err(e) => {
164 - crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, "cli_upload_rejected").await;
164 + crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "cli_upload_rejected").await;
165 165 return Err(e);
166 166 }
167 167 };
@@ -169,7 +169,7 @@ pub(super) async fn confirm_upload(
169 169 // Reject unsupported file types BEFORE any side effect — same ordering rule
170 170 // as the web upload handlers (see routes/storage/mod.rs::commit_upload).
171 171 if !matches!(file_type, FileType::Audio | FileType::Download | FileType::Video) {
172 - crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, "cli_upload_rejected").await;
172 + crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "cli_upload_rejected").await;
173 173 return Err(AppError::BadRequest(
174 174 "CLI upload only supports audio, video, and download file types".to_string(),
175 175 ));
@@ -178,7 +178,7 @@ pub(super) async fn confirm_upload(
178 178 // Increment storage BEFORE writing the DB record (if quota exceeded, the
179 179 // S3 object is cleaned up and no DB record is created).
180 180 if let Err(e) = db::creator_tiers::try_increment_storage(&state.db, req.user_id, file_size_bytes, max_storage).await {
181 - crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, "cli_upload_rejected").await;
181 + crate::routes::storage::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "cli_upload_rejected").await;
182 182 return Err(e);
183 183 }
184 184
M server/src/routes/storage/gallery.rs +4 -4
@@ -235,7 +235,7 @@ pub(super) async fn gallery_confirm(
235 235 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
236 236 })?;
237 237 if file_size_bytes as u64 > FileType::Cover.max_size() {
238 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "gallery_upload_rejected").await;
238 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
239 239 return Err(AppError::BadRequest(format!(
240 240 "File exceeds maximum size of {} MB",
241 241 FileType::Cover.max_size() / (1024 * 1024)
@@ -248,7 +248,7 @@ pub(super) async fn gallery_confirm(
248 248 GalleryTarget::Project => db::gallery_images::count_for_project(&state.db, ProjectId::from(req.target_id)).await?,
249 249 };
250 250 if count >= db::gallery_images::MAX_GALLERY_IMAGES {
251 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "gallery_upload_rejected").await;
251 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
252 252 return Err(AppError::BadRequest(format!(
253 253 "Gallery is full (max {} images)",
254 254 db::gallery_images::MAX_GALLERY_IMAGES
@@ -258,7 +258,7 @@ pub(super) async fn gallery_confirm(
258 258 let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Cover, file_size_bytes).await {
259 259 Ok(max) => max,
260 260 Err(e) => {
261 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "gallery_upload_rejected").await;
261 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_upload_rejected").await;
262 262 return Err(e);
263 263 }
264 264 };
@@ -363,7 +363,7 @@ pub(super) async fn gallery_confirm(
363 363 }
364 364 Ok(GalleryOutcome::Inserted(id)) => id,
365 365 Err(e) => {
366 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "gallery_image_insert_failed").await;
366 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "gallery_image_insert_failed").await;
367 367 return Err(e);
368 368 }
369 369 };
M server/src/routes/storage/images.rs +15 -11
@@ -140,7 +140,7 @@ pub(super) async fn project_image_confirm(
140 140 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
141 141 })?;
142 142 if file_size_bytes as u64 > FileType::Cover.max_size() {
143 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "image_upload_rejected").await;
143 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
144 144 return Err(AppError::BadRequest(format!(
145 145 "File exceeds maximum size of {} MB",
146 146 FileType::Cover.max_size() / (1024 * 1024)
@@ -174,7 +174,7 @@ pub(super) async fn project_image_confirm(
174 174 let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Cover, file_size_bytes).await {
175 175 Ok(max) => max,
176 176 Err(e) => {
177 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "image_upload_rejected").await;
177 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
178 178 return Err(e);
179 179 }
180 180 };
@@ -207,7 +207,7 @@ pub(super) async fn project_image_confirm(
207 207 Err(e) => {
208 208 // S3 probe failed (transient). Refuse to write — letting a probe failure
209 209 // silently over-count storage on every replace is the bug this branch exists to prevent.
210 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "image_upload_rejected").await;
210 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
211 211 tracing::warn!(key = %old_key, error = ?e, "S3 probe failed during project image replace");
212 212 return Err(AppError::ServiceUnavailable(
213 213 "Could not verify previous image. Please try again.".to_string(),
@@ -262,11 +262,11 @@ pub(super) async fn project_image_confirm(
262 262 match committed {
263 263 Err(e) => {
264 264 // tx rolled back — counter unchanged. Orphan-queue the new key for cleanup.
265 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "project_image_update_failed").await;
265 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "project_image_update_failed").await;
266 266 return Err(e);
267 267 }
268 268 Ok(false) => {
269 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "project_image_update_failed").await;
269 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "project_image_update_failed").await;
270 270 return Err(AppError::BadRequest(
271 271 "Project was modified concurrently. Please try uploading again.".to_string(),
272 272 ));
@@ -378,8 +378,12 @@ pub(super) async fn item_image_confirm(
378 378 return Err(AppError::Forbidden);
379 379 }
380 380
381 - // Validate S3 key belongs to this user + item (prevent cross-user file reference)
382 - let expected_prefix = format!("{}/{}/", user.id, req.item_id);
381 + // Validate the key belongs to this user + item AND sits under the `cover/`
382 + // segment that `generate_key(.., Cover, ..)` mints — not just `{user}/{item}/`.
383 + // The loose prefix let a confirm point cover_s3_key at any object the user
384 + // owns under the item (e.g. a `download/` payload), asymmetric with the
385 + // version path's full `{user}/{item}/download/{version}/` pin (Run 11 Storage LOW).
386 + let expected_prefix = format!("{}/{}/{}/", user.id, req.item_id, FileType::Cover.as_str());
383 387 if !req.s3_key.starts_with(&expected_prefix) {
384 388 return Err(AppError::BadRequest(
385 389 "Invalid upload key".to_string(),
@@ -392,7 +396,7 @@ pub(super) async fn item_image_confirm(
392 396 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
393 397 })?;
394 398 if file_size_bytes as u64 > FileType::Cover.max_size() {
395 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "image_upload_rejected").await;
399 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
396 400 return Err(AppError::BadRequest(format!(
397 401 "File exceeds maximum size of {} MB",
398 402 FileType::Cover.max_size() / (1024 * 1024)
@@ -422,7 +426,7 @@ pub(super) async fn item_image_confirm(
422 426 let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Cover, file_size_bytes).await {
423 427 Ok(max) => max,
424 428 Err(e) => {
425 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "image_upload_rejected").await;
429 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "image_upload_rejected").await;
426 430 return Err(e);
427 431 }
428 432 };
@@ -484,11 +488,11 @@ pub(super) async fn item_image_confirm(
484 488
485 489 match committed {
486 490 Err(e) => {
487 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_image_update_failed").await;
491 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_image_update_failed").await;
488 492 return Err(e);
489 493 }
490 494 Ok(false) => {
491 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_image_update_failed").await;
495 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_image_update_failed").await;
492 496 return Err(AppError::BadRequest(
493 497 "Item was modified concurrently. Please try uploading again.".to_string(),
494 498 ));
M server/src/routes/storage/media.rs +4 -4
@@ -227,7 +227,7 @@ pub(super) async fn media_confirm(
227 227 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
228 228 })?;
229 229 if file_size_bytes as u64 > file_type.max_size() {
230 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "media_upload_rejected").await;
230 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
231 231 return Err(AppError::BadRequest(format!(
232 232 "File exceeds maximum size of {} MB",
233 233 file_type.max_size() / (1024 * 1024)
@@ -263,7 +263,7 @@ pub(super) async fn media_confirm(
263 263 _ => false,
264 264 };
265 265 if mismatch {
266 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "media_content_type_mismatch").await;
266 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_content_type_mismatch").await;
267 267 let sniffed = match detected {
268 268 Some(infer::MatcherType::Image) => "image",
269 269 Some(infer::MatcherType::Video) => "video",
@@ -283,7 +283,7 @@ pub(super) async fn media_confirm(
283 283 {
284 284 Ok(max) => max,
285 285 Err(e) => {
286 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "media_upload_rejected").await;
286 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
287 287 return Err(e);
288 288 }
289 289 };
@@ -361,7 +361,7 @@ pub(super) async fn media_confirm(
361 361 }
362 362 // Any other failure: the tx rolled back and no row references this
363 363 // freshly-uploaded object, so it's a genuine orphan — clean it up.
364 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "media_upload_rejected").await;
364 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "media_upload_rejected").await;
365 365 return Err(e);
366 366 }
367 367 };
M server/src/routes/storage/mod.rs +8 -3
@@ -76,15 +76,20 @@ pub(crate) fn validate_declared_upload_size(
76 76 /// routes through here. The queue worker applies the `is_s3_key_live` guard
77 77 /// before deleting, so enqueuing a key a live row still references is safe (it
78 78 /// is skipped), and a transient failure is retried rather than leaking.
79 - pub(crate) async fn enqueue_s3_orphan(pool: &sqlx::PgPool, s3_key: &str, source: &'static str) {
79 + pub(crate) async fn enqueue_s3_orphan(
80 + pool: &sqlx::PgPool,
81 + s3_key: &str,
82 + bucket: crate::storage::S3Bucket,
83 + source: &'static str,
84 + ) {
80 85 if let Err(e) = db::pending_s3_deletions::enqueue_deletions(
81 86 pool,
82 - &[(s3_key.to_string(), "main".to_string())],
87 + &[(s3_key.to_string(), bucket.as_str().to_string())],
83 88 source,
84 89 )
85 90 .await
86 91 {
87 - tracing::warn!(error = ?e, key = %s3_key, source = %source, "failed to enqueue orphan S3 key");
92 + tracing::warn!(error = ?e, key = %s3_key, bucket = %bucket.as_str(), source = %source, "failed to enqueue orphan S3 key");
88 93 }
89 94 }
90 95
M server/src/routes/storage/uploads.rs +5 -5
@@ -160,7 +160,7 @@ pub(super) async fn confirm_upload(
160 160 AppError::BadRequest("Could not determine file size. Please try uploading again.".to_string())
161 161 })?;
162 162 if file_size_bytes as u64 > file_type.max_size() {
163 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_upload_rejected").await;
163 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
164 164 let limit_mb = file_type.max_size() / (1024 * 1024);
165 165 let file_mb = file_size_bytes as u64 / (1024 * 1024);
166 166 return Err(AppError::FileTooLarge(format!(
@@ -173,7 +173,7 @@ pub(super) async fn confirm_upload(
173 173 let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, file_type, file_size_bytes).await {
174 174 Ok(max) => max,
175 175 Err(e) => {
176 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_upload_rejected").await;
176 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
177 177 return Err(e);
178 178 }
179 179 };
@@ -187,7 +187,7 @@ pub(super) async fn confirm_upload(
187 187 // which only /api/items/image/confirm writes — the old generic two-column
188 188 // path left it NULL and rendered an invisible cover (Run #13 SERIOUS).
189 189 if let crate::storage::GenericItemConfirm::UseRoute(route) = file_type.generic_item_confirm() {
190 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_upload_rejected").await;
190 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_rejected").await;
191 191 return Err(AppError::BadRequest(format!(
192 192 "This file type isn't confirmed here — use {route}."
193 193 )));
@@ -291,7 +291,7 @@ pub(super) async fn confirm_upload(
291 291 // filling in between), so a blind delete could destroy the live
292 292 // object the winner points at. Route through the orphan queue; its
293 293 // `is_s3_key_live` check skips any key a row still references.
294 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_confirm_failed").await;
294 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_confirm_failed").await;
295 295 return Err(e);
296 296 }
297 297 Ok(0) => {
@@ -301,7 +301,7 @@ pub(super) async fn confirm_upload(
301 301 // observed). Either way the tx rolled back, nothing was charged;
302 302 // route the now-unreferenced object through the orphan queue so the
303 303 // reaper still cleans it.
304 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "item_upload_target_missing").await;
304 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "item_upload_target_missing").await;
305 305 return Err(AppError::BadRequest(
306 306 "Item was modified concurrently. Please try uploading again.".to_string(),
307 307 ));
M server/src/routes/storage/versions.rs +4 -4
@@ -146,7 +146,7 @@ pub(super) async fn version_confirm_upload(
146 146 AppError::BadRequest("Upload not found. Please try uploading again.".to_string())
147 147 })?;
148 148 if file_size_bytes as u64 > FileType::Download.max_size() {
149 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "version_upload_rejected").await;
149 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_upload_rejected").await;
150 150 let limit_mb = FileType::Download.max_size() / (1024 * 1024);
151 151 let file_mb = file_size_bytes as u64 / (1024 * 1024);
152 152 return Err(AppError::FileTooLarge(format!(
@@ -159,7 +159,7 @@ pub(super) async fn version_confirm_upload(
159 159 let max_storage = match db::creator_tiers::check_upload_allowed(&state.db, user.id, FileType::Download, file_size_bytes).await {
160 160 Ok(max) => max,
161 161 Err(e) => {
162 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "version_upload_rejected").await;
162 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_upload_rejected").await;
163 163 return Err(e);
164 164 }
165 165 };
@@ -235,7 +235,7 @@ pub(super) async fn version_confirm_upload(
235 235 // would then destroy the live object the winning confirm points at.
236 236 // Route through the orphan queue, whose `is_s3_key_live` check skips
237 237 // any key a row still references.
238 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "version_confirm_failed").await;
238 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_confirm_failed").await;
239 239 return Err(e);
240 240 }
241 241 Ok(false) => {
@@ -243,7 +243,7 @@ pub(super) async fn version_confirm_upload(
243 243 // s3_key out from under us. If it committed THIS key, a direct
244 244 // delete would 404 every fan download of the version the winner
245 245 // just published. The orphan queue's liveness check is the guard.
246 - super::enqueue_s3_orphan(&state.db, &req.s3_key, "version_confirm_lost_race").await;
246 + super::enqueue_s3_orphan(&state.db, &req.s3_key, crate::storage::S3Bucket::Main, "version_confirm_lost_race").await;
247 247 return Err(AppError::BadRequest(
248 248 "Version was modified concurrently. Please try uploading again.".to_string(),
249 249 ));
M server/src/scheduler/cleanup.rs +67 -29
@@ -97,14 +97,16 @@ async fn cleanup_user_s3_and_delete(state: &AppState, user_id: db::UserId, event
97 97 .unwrap_or_default();
98 98
99 99 let user_prefix = format!("{user_id}/");
100 + let main = crate::storage::S3Bucket::Main.as_str().to_string();
101 + let synckit = crate::storage::S3Bucket::Synckit.as_str().to_string();
100 102 let mut keys: Vec<(String, String)> = Vec::new();
101 - keys.push((user_prefix.clone(), "main".to_string()));
103 + keys.push((user_prefix.clone(), main.clone()));
102 104 for pid in &project_ids {
103 - keys.push((format!("projects/{pid}/"), "main".to_string()));
105 + keys.push((format!("projects/{pid}/"), main.clone()));
104 106 }
105 107 for app in &sync_apps {
106 - keys.push((format!("{}/", app.id), "synckit".to_string()));
107 - keys.push((format!("ota/{}/", app.id), "synckit".to_string()));
108 + keys.push((format!("{}/", app.id), synckit.clone()));
109 + keys.push((format!("ota/{}/", app.id), synckit.clone()));
108 110 }
109 111
110 112 // Enqueue all keys before any destructive work
@@ -323,7 +325,7 @@ pub(super) async fn purge_expired_deleted_items(state: &AppState) {
323 325 match db::items::get_expired_deleted_item_s3_keys(&state.db).await {
324 326 Ok(keys) => {
325 327 for key in &keys {
326 - all_s3_keys.push((key.clone(), "main".to_string()));
328 + all_s3_keys.push((key.clone(), crate::storage::S3Bucket::Main.as_str().to_string()));
327 329 }
328 330 }
329 331 Err(e) => {
@@ -334,7 +336,7 @@ pub(super) async fn purge_expired_deleted_items(state: &AppState) {
334 336 match db::items::get_expired_deleted_item_version_s3_keys(&state.db).await {
335 337 Ok(keys) => {
336 338 for key in &keys {
337 - all_s3_keys.push((key.clone(), "main".to_string()));
339 + all_s3_keys.push((key.clone(), crate::storage::S3Bucket::Main.as_str().to_string()));
338 340 }
339 341 }
340 342 Err(e) => {
@@ -347,7 +349,7 @@ pub(super) async fn purge_expired_deleted_items(state: &AppState) {
347 349 match db::gallery_images::s3_keys_for_expired_purged_items(&state.db).await {
348 350 Ok(keys) => {
349 351 for key in &keys {
350 - all_s3_keys.push((key.clone(), "main".to_string()));
352 + all_s3_keys.push((key.clone(), crate::storage::S3Bucket::Main.as_str().to_string()));
351 353 }
352 354 }
353 355 Err(e) => {
@@ -484,9 +486,9 @@ pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
484 486 let mut failed_keys: Vec<(String, String)> = Vec::new();
485 487
486 488 for (s3_key, bucket) in &stale {
487 - let s3_client = match bucket.as_str() {
488 - "synckit" => state.synckit_s3.as_ref(),
489 - _ => state.s3.as_ref(),
489 + let s3_client = match crate::storage::S3Bucket::from_db_str(bucket) {
490 + crate::storage::S3Bucket::Synckit => state.synckit_s3.as_ref(),
491 + crate::storage::S3Bucket::Main => state.s3.as_ref(),
490 492 };
491 493 if let Some(s3) = s3_client {
492 494 // Route through the guarded funnel: if a confirm already reclaimed
@@ -589,28 +591,64 @@ pub(super) async fn retry_pending_s3_deletions(state: &AppState) {
589 591 "S3 deletion stuck after 5+ attempts");
590 592 }
591 593
592 - let s3 = if row.bucket == "synckit" {
593 - state.synckit_s3.as_ref()
594 - } else {
595 - state.s3.as_ref()
594 + let s3 = match crate::storage::S3Bucket::from_db_str(&row.bucket) {
595 + crate::storage::S3Bucket::Synckit => state.synckit_s3.as_ref(),
596 + crate::storage::S3Bucket::Main => state.s3.as_ref(),
596 597 };
597 598
598 599 if let Some(s3) = s3 {
599 600 if row.s3_key.ends_with('/') {
600 - // Prefix delete (user-scoped cascade cleanup). Bypasses the
601 - // per-key live-check by design, but carries its own guard: a bare
602 - // `{user_id}/` prefix wipes a creator's ENTIRE storage and is only
603 - // ever enqueued by account cleanup (which CASCADE-deletes the user
604 - // row in the same pass). A STILL-LIVE user here means something
605 - // wrongly enqueued a prefix delete against a live account — refuse
606 - // and let the row climb toward dead-letter triage rather than
607 - // silently nuking live files.
608 - if row.bucket == "main"
609 - && let Some(uid) = row.s3_key.strip_suffix('/').and_then(|s| s.parse::<db::UserId>().ok())
610 - && matches!(db::users::get_user_by_id(&state.db, uid).await, Ok(Some(_)))
611 - {
612 - tracing::error!(s3_key = %row.s3_key, user_id = %uid,
613 - "refusing user-prefix S3 delete: user still exists — parking for dead-letter triage instead of wiping live storage");
601 + // Prefix delete (account-cascade cleanup). Bypasses the per-key
602 + // live-check by design, but carries a liveness guard symmetric
603 + // across buckets: a `{user_id}/` (main) or `{app_id}/` /
604 + // `ota/{app_id}/` (synckit) prefix wipes an entire creator's or
605 + // app's storage and is only ever enqueued by cleanup that deletes
606 + // the owning row in the same pass. A STILL-LIVE owner means
607 + // something wrongly enqueued a wipe — refuse and let the row climb
608 + // toward dead-letter triage rather than nuking live files. The
609 + // synckit branch was previously unguarded (ultra-fuzz Run 11 Storage LOW).
610 + let live_owner: Option<String> =
611 + match crate::storage::S3Bucket::from_db_str(&row.bucket) {
612 + crate::storage::S3Bucket::Main => {
613 + match row
614 + .s3_key
615 + .strip_suffix('/')
616 + .and_then(|s| s.parse::<db::UserId>().ok())
617 + {
618 + Some(uid)
619 + if matches!(
620 + db::users::get_user_by_id(&state.db, uid).await,
621 + Ok(Some(_))
622 + ) =>
623 + {
624 + Some(format!("user {uid}"))
625 + }
626 + _ => None,
627 + }
628 + }
629 + crate::storage::S3Bucket::Synckit => {
630 + let app_seg =
631 + row.s3_key.strip_prefix("ota/").unwrap_or(&row.s3_key);
632 + match app_seg
633 + .strip_suffix('/')
634 + .and_then(|s| s.parse::<db::SyncAppId>().ok())
635 + {
636 + Some(app_id)
637 + if matches!(
638 + db::synckit::get_sync_app_by_id(&state.db, app_id)
639 + .await,
640 + Ok(Some(_))
641 + ) =>
642 + {
643 + Some(format!("sync app {app_id}"))
644 + }
645 + _ => None,
646 + }
647 + }
648 + };
649 + if let Some(owner) = live_owner {
650 + tracing::error!(s3_key = %row.s3_key, %owner,
651 + "refusing prefix S3 delete: owner still exists — parking for dead-letter triage instead of wiping live storage");
614 652 continue;
615 653 }
616 654 match s3.delete_prefix(&S3DeleteAuthority::new(), &row.s3_key).await {
@@ -682,7 +720,7 @@ pub async fn drain_pending_s3_deletions_for_test(
682 720 let mut deleted = 0usize;
683 721 for row in &stale {
684 722 // Storage tests only exercise the main bucket; synckit needs its own s3.
685 - if row.bucket != "main" {
723 + if crate::storage::S3Bucket::from_db_str(&row.bucket) != crate::storage::S3Bucket::Main {
686 724 continue;
687 725 }
688 726 match delete_orphan_key_guarded(pool, s3, &row.bucket, &row.s3_key).await {
M server/src/storage.rs +36
@@ -300,6 +300,42 @@ impl S3DeleteAuthority {
300 300 }
301 301 }
302 302
303 + /// Which configured S3 backend an object lives in.
304 + ///
305 + /// The delete *verb* is type-sealed by [`S3DeleteAuthority`]; this seals the
306 + /// bucket *noun*. The `pending_s3_deletions` queue stores the bucket as text, and
307 + /// the deletion worker dispatches between the main and SyncKit S3 clients on that
308 + /// text. This enum is the single source of truth for the `"main"`/`"synckit"`
309 + /// spellings so an orphan-enqueue can't silently mis-tag a SyncKit object as
310 + /// `main` (where the worker would delete it against the wrong client and leak it
311 + /// forever) — `enqueue_s3_orphan` now requires an `S3Bucket`, not a bare string
312 + /// (ultra-fuzz Run 11 Storage surprise).
313 + #[derive(Clone, Copy, Debug, PartialEq, Eq)]
314 + pub enum S3Bucket {
315 + Main,
316 + Synckit,
317 + }
318 +
319 + impl S3Bucket {
320 + /// The stored/text spelling for the deletion queue.
321 + pub fn as_str(self) -> &'static str {
322 + match self {
323 + S3Bucket::Main => "main",
324 + S3Bucket::Synckit => "synckit",
325 + }
326 + }
327 +
328 + /// Parse a bucket tag read back from the queue. Unknown/legacy values map to
329 + /// `Main` (the historical default), so a garbled row is still reaped against
330 + /// a backend rather than wedging the queue.
331 + pub fn from_db_str(s: &str) -> Self {
332 + match s {
333 + "synckit" => S3Bucket::Synckit,
334 + _ => S3Bucket::Main,
335 + }
336 + }
337 + }
338 +
303 339 /// Abstract storage backend — implemented by `S3Client` (production) and
304 340 /// `InMemoryStorage` (tests). Routes access storage through this trait.
305 341 #[async_trait::async_trait]
M shared/s3-storage/src/lib.rs +64 -25
@@ -208,10 +208,20 @@ impl S3Client {
208 208 if keys.is_empty() {
209 209 return Ok(Vec::new());
210 210 }
211 - let objects: Vec<ObjectIdentifier> = keys
212 - .iter()
213 - .filter_map(|k| ObjectIdentifier::builder().key(k).build().ok())
214 - .collect();
211 + // A key that won't build an ObjectIdentifier is reported as a failure
212 + // rather than silently dropped — otherwise the caller believes it was
213 + // deleted and the object leaks forever (ultra-fuzz Run 11 Storage MED).
214 + let mut failures: Vec<(String, String)> = Vec::new();
215 + let mut objects: Vec<ObjectIdentifier> = Vec::with_capacity(keys.len());
216 + for k in keys {
217 + match ObjectIdentifier::builder().key(k).build() {
218 + Ok(o) => objects.push(o),
219 + Err(e) => failures.push((k.clone(), format!("malformed key: {e}"))),
220 + }
221 + }
222 + if objects.is_empty() {
223 + return Ok(failures);
224 + }
215 225 let delete = Delete::builder()
216 226 .set_objects(Some(objects))
217 227 .quiet(true)
@@ -225,16 +235,11 @@ impl S3Client {
225 235 .send()
226 236 .await
227 237 .map_err(|e| format!("S3 delete_objects failed: {e}"))?;
228 - let failures = resp
229 - .errors
230 - .unwrap_or_default()
231 - .into_iter()
232 - .filter_map(|err| {
233 - let key = err.key?;
234 - let msg = err.message.unwrap_or_else(|| "<no message>".into());
235 - Some((key, msg))
236 - })
237 - .collect();
238 + failures.extend(resp.errors.unwrap_or_default().into_iter().filter_map(|err| {
239 + let key = err.key?;
240 + let msg = err.message.unwrap_or_else(|| "<no message>".into());
241 + Some((key, msg))
242 + }));
238 243 Ok(failures)
239 244 }
240 245
@@ -483,9 +488,15 @@ impl S3Client {
483 488
484 489 let mut part_number: i32 = 1;
485 490 let mut completed_parts: Vec<CompletedPart> = Vec::new();
486 - let mut buf = vec![0u8; part_size];
487 491
488 492 loop {
493 + // One owned buffer per part, frozen into `Bytes` so each retry attempt
494 + // clones a refcount instead of re-copying the part. Previously the
495 + // body was `buf[..n].to_vec()` *inside* the retry loop, so every part
496 + // (and every retry) paid a full part_size heap copy on top of the
497 + // resident staging buffer — ~2x part_size live (ultra-fuzz Run 11
498 + // Storage HIGH). `Bytes::from(Vec)` takes ownership without copying.
499 + let mut buf = vec![0u8; part_size];
489 500 let mut bytes_read = 0;
490 501 // Fill the buffer completely (or until EOF)
491 502 while bytes_read < part_size {
@@ -500,6 +511,9 @@ impl S3Client {
500 511 break;
501 512 }
502 513
514 + buf.truncate(bytes_read);
515 + let part: bytes::Bytes = buf.into();
516 +
503 517 // Retry the part upload up to 3 times on transient failures.
504 518 // S3 part uploads can flake on network blips; aborting the
505 519 // whole multipart upload because of one timeout means the
@@ -510,7 +524,7 @@ impl S3Client {
510 524 let mut attempt: u32 = 0;
511 525 let resp = loop {
512 526 attempt += 1;
513 - let body = aws_sdk_s3::primitives::ByteStream::from(buf[..bytes_read].to_vec());
527 + let body = aws_sdk_s3::primitives::ByteStream::from(part.clone());
514 528 match self
515 529 .client
516 530 .upload_part()
@@ -560,15 +574,40 @@ impl S3Client {
560 574 .set_parts(Some(completed_parts))
561 575 .build();
562 576
563 - self.client
564 - .complete_multipart_upload()
565 - .bucket(&self.bucket)
566 - .key(key)
567 - .upload_id(upload_id)
568 - .multipart_upload(completed)
569 - .send()
570 - .await
571 - .map_err(|e| format!("S3 complete multipart upload failed: {e}"))?;
577 + // Retry completion on transient failures, same as the part uploads.
578 + // Completion is the most expensive call to lose — every part is already
579 + // uploaded and paid for, so a single flake here would otherwise orphan
580 + // the whole multipart upload and force a restart from byte 0 (ultra-fuzz
581 + // Run 11 Storage HIGH).
582 + let mut attempt: u32 = 0;
583 + loop {
584 + attempt += 1;
585 + match self
586 + .client
587 + .complete_multipart_upload()
588 + .bucket(&self.bucket)
589 + .key(key)
590 + .upload_id(upload_id)
591 + .multipart_upload(completed.clone())
592 + .send()
593 + .await
594 + {
595 + Ok(_) => break,
596 + Err(e) if attempt < 3 => {
597 + let delay_ms = 200u64 * (1u64 << ((attempt - 1) * 2));
598 + tracing::warn!(
599 + attempt, delay_ms, error = ?e,
600 + "S3 complete_multipart_upload transient failure, retrying"
601 + );
602 + tokio::time::sleep(std::time::Duration::from_millis(delay_ms)).await;
603 + }
604 + Err(e) => {
605 + return Err(format!(
606 + "S3 complete multipart upload failed after retries: {e}"
607 + ));
608 + }
609 + }
610 + }
572 611
573 612 Ok(())
574 613 }