Skip to main content

max / makenotwork

server: orphan reaper hands transient failures to dead-letter ladder (ultra-fuzz Run 2 Storage SERIOUS) SERIOUS: the orphaned-upload reaper cleared the pending_uploads tracking row even when the guarded S3 delete returned Failed — a transient S3 error (timeout, 503, throttle) on a genuine orphan permanently leaked the object, untracked by any queue. Failed deletes now hand off to pending_s3_deletions (the same retry + dead-letter ladder the sibling deletion worker already uses) BEFORE the tracking row is cleared; if the handoff itself fails, the row is left for the next reaper tick rather than dropped. MINOR: delete_objects (trait default + S3Client batch override) returned Ok even when every key in a batch failed; a total failure now returns Err so a caller without its own safety net isn't told a wholesale failure succeeded. Partial failures stay logged (callers pre-enqueue to pending_s3_deletions). The gated-kind quarantine-purge retry MINOR is deferred: it's a rare malware path that already opens a WAM ticket, and inline retry-with-backoff in the scan worker is risk disproportionate to the gain. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-06-23 16:48 UTC
Commit: 766f40f2073b59c087b9f30c7119a8a2a9dee520
Parent: f66d659
2 files changed, +49 insertions, -3 deletions
M server/src/scheduler/cleanup.rs +31 -2
@@ -469,6 +469,12 @@ pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
469 469 // Carry (key, bucket) pairs: the deletion is bucket-scoped so a key present
470 470 // in two buckets only clears the record for the bucket we reaped (B7).
471 471 let mut keys_to_delete: Vec<(String, String)> = Vec::with_capacity(stale.len());
472 + // Transient S3 delete failures handed off to the durable deletion queue
473 + // (which has a retry + dead-letter backstop) instead of dropping the only
474 + // tracking row — clearing it on a transient failure permanently leaked the
475 + // object (Run #2 Storage SERIOUS; the sibling retry worker already routes
476 + // failures into this ladder).
477 + let mut failed_keys: Vec<(String, String)> = Vec::new();
472 478
473 479 for (s3_key, bucket) in &stale {
474 480 let s3_client = match bucket.as_str() {
@@ -486,8 +492,9 @@ pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
486 492 }
487 493 // Live row owns the key now: clear the stale record, keep object.
488 494 GuardedDelete::SkippedLive => keys_to_delete.push((s3_key.clone(), bucket.clone())),
489 - // Delete failed: still clear the record so we don't retry forever.
490 - GuardedDelete::Failed => keys_to_delete.push((s3_key.clone(), bucket.clone())),
495 + // Delete failed (transient): hand off to the durable deletion
496 + // queue rather than dropping the tracking row and leaking.
497 + GuardedDelete::Failed => failed_keys.push((s3_key.clone(), bucket.clone())),
491 498 }
492 499 } else {
493 500 // S3 not configured for this bucket; remove the DB record anyway
@@ -495,6 +502,28 @@ pub(super) async fn cleanup_orphaned_uploads(state: &AppState) {
495 502 }
496 503 }
497 504
505 + // Enqueue transient failures into the durable deletion queue BEFORE clearing
506 + // their tracking rows, so a key is never dropped by both. If the handoff
507 + // itself fails, leave the pending_uploads rows for the next reaper tick
508 + // rather than leaking.
509 + if !failed_keys.is_empty() {
510 + match db::pending_s3_deletions::enqueue_deletions(&state.db, &failed_keys, "orphan-reaper-retry").await {
511 + Ok(()) => {
512 + tracing::warn!(
513 + count = failed_keys.len(),
514 + "orphan reaper: S3 delete failed; handed off to durable deletion queue for retry"
515 + );
516 + keys_to_delete.extend(failed_keys);
517 + }
518 + Err(e) => {
519 + tracing::error!(
520 + error = ?e, count = failed_keys.len(),
521 + "orphan reaper: could not enqueue transient failures to deletion queue; leaving tracking rows for next tick"
522 + );
523 + }
524 + }
525 + }
526 +
498 527 if !keys_to_delete.is_empty()
499 528 && let Err(e) = db::pending_uploads::delete_pending_uploads(&state.db, &keys_to_delete).await
500 529 {
M server/src/storage.rs +18 -1
@@ -330,11 +330,19 @@ pub trait StorageBackend: Send + Sync {
330 330 /// (up to 1000 keys/call). Default loops `delete_object` so test backends
331 331 /// don't have to implement it, but production should override.
332 332 async fn delete_objects(&self, auth: &S3DeleteAuthority, keys: &[S3Key]) -> Result<()> {
333 + let mut failed = 0usize;
333 334 for k in keys {
334 335 if let Err(e) = self.delete_object(auth, k).await {
336 + failed += 1;
335 337 tracing::warn!(key = %k, error = ?e, "delete_objects: per-key delete failed");
336 338 }
337 339 }
340 + // Don't report success when every key failed — a total failure must
341 + // surface so the caller can fall back (Run #2 Storage MINOR). Partial
342 + // failures stay logged; callers pre-enqueue to pending_s3_deletions.
343 + if !keys.is_empty() && failed == keys.len() {
344 + return Err(AppError::Storage(format!("delete_objects: all {failed} keys failed")));
345 + }
338 346 Ok(())
339 347 }
340 348 /// Delete all objects under a key prefix. Default logs a warning (no-op).
@@ -664,9 +672,18 @@ impl S3Client {
664 672 let chunk: Vec<String> = chunk.iter().map(|k| k.as_str().to_string()).collect();
665 673 match self.inner.delete_objects(&chunk).await {
666 674 Ok(failures) => {
667 - for (k, msg) in failures {
675 + for (k, msg) in &failures {
668 676 tracing::warn!(key = %k, error = %msg, "S3 delete_objects: key-level failure");
669 677 }
678 + // A whole-batch failure must not read as success (Run #2
679 + // Storage MINOR); partial failures stay logged and the
680 + // pending_s3_deletions queue is the retry net.
681 + if !chunk.is_empty() && failures.len() == chunk.len() {
682 + return Err(AppError::Storage(format!(
683 + "S3 delete_objects: all {} keys in batch failed",
684 + chunk.len()
685 + )));
686 + }
670 687 }
671 688 Err(e) => return Err(AppError::Storage(e)),
672 689 }