max / makenotwork
- Co-Authored-By
- Claude Opus 5 (1M context) <noreply@anthropic.com>
5 files changed,
+449 insertions,
-407 deletions
| @@ -51,7 +51,9 @@ | |||
| 51 | 51 | mod embeds; | |
| 52 | 52 | mod enum_drift; | |
| 53 | 53 | mod exports; | |
| 54 | - | mod failure_paths; | |
| 54 | + | mod failure_paths_checkout; | |
| 55 | + | mod failure_paths_scan; | |
| 56 | + | mod failure_paths_storage; | |
| 55 | 57 | mod fan_plus; | |
| 56 | 58 | mod fingerprinting; | |
| 57 | 59 | mod follows; |
| @@ -336,11 +336,7 @@ | |||
| 336 | 336 | &format!("title={title}&item_type=text&is_public=true&price_cents=0"), | |
| 337 | 337 | ) | |
| 338 | 338 | .await; | |
| 339 | - | assert!( | |
| 340 | - | resp.status.is_success(), | |
| 341 | - | "create item failed: {}", | |
| 342 | - | resp.text | |
| 343 | - | ); | |
| 339 | + | assert_eq!(resp.status, 200, "create item failed: {}", resp.text); | |
| 344 | 340 | } | |
| 345 | 341 | ||
| 346 | 342 | // Price the project itself, which is what puts the page behind the paywall. | |
| @@ -351,8 +347,8 @@ | |||
| 351 | 347 | r#"{"pricing_model": "buy_once", "price_dollars": 9.0}"#, | |
| 352 | 348 | ) | |
| 353 | 349 | .await; | |
| 354 | - | assert!( | |
| 355 | - | resp.status.is_success(), | |
| 350 | + | assert_eq!( | |
| 351 | + | resp.status, 200, | |
| 356 | 352 | "set project pricing failed: {}", | |
| 357 | 353 | resp.text | |
| 358 | 354 | ); |
R
server/tests/workflows/failure_paths.rs → server/tests/workflows/failure_paths_checkout.rs
+8
-399
| @@ -1,110 +1,25 @@ | |||
| 1 | - | //! Negative paths: what the server does when a dependency fails. | |
| 1 | + | //! Negative paths: what the server does when Stripe or email fails. | |
| 2 | + | //! | |
| 3 | + | //! Split out of the former `failure_paths.rs` on 2026-08-05, which reached 868 | |
| 4 | + | //! lines and tripped the oversized-module ratchet in `tests/test_hygiene.rs`. | |
| 5 | + | //! The split is by the dependency that fails, which is also how you look these | |
| 6 | + | //! up: the money path and the mail that follows it share a setup and a blast radius. | |
| 2 | 7 | //! | |
| 3 | 8 | //! These tests exist because the mocks used to be infallible, so the retry and | |
| 4 | 9 | //! compensation machinery the server carries had no test that could reach it. | |
| 5 | 10 | //! Retry logic no test can enter is worse than none, because it reads as | |
| 6 | - | //! handled. Each test here installs a failure policy on a mock (see | |
| 11 | + | //! handled. Each test installs a failure policy on a mock (see | |
| 7 | 12 | //! `harness::faults`) and asserts the compensating behaviour, not just that the | |
| 8 | 13 | //! request failed. | |
| 9 | 14 | //! | |
| 10 | 15 | //! Rationale: wiki `testing-posture`, the "absent oracle" section. | |
| 11 | 16 | ||
| 12 | 17 | use crate::harness::TestHarness; | |
| 13 | - | use crate::harness::faults::{email_unavailable, storage_unavailable, stripe_unavailable}; | |
| 18 | + | use crate::harness::faults::{email_unavailable, stripe_unavailable}; | |
| 14 | 19 | use makenotwork::db; | |
| 15 | - | use makenotwork::storage::StorageBackend; | |
| 16 | 20 | use serde_json::Value; | |
| 17 | 21 | use std::collections::HashMap; | |
| 18 | 22 | ||
| 19 | - | // The durable S3 deletion queue | |
| 20 | - | ||
| 21 | - | /// Count rows still queued for deletion of `key`. | |
| 22 | - | async fn queued_deletions(h: &TestHarness, key: &str) -> i64 { | |
| 23 | - | sqlx::query_scalar("SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1") | |
| 24 | - | .bind(key) | |
| 25 | - | .fetch_one(&h.db) | |
| 26 | - | .await | |
| 27 | - | .unwrap() | |
| 28 | - | } | |
| 29 | - | ||
| 30 | - | /// A delete that fails must leave the row queued. Dequeuing it would orphan the | |
| 31 | - | /// S3 object with no durable record, which is the leak the queue exists to | |
| 32 | - | /// prevent. | |
| 33 | - | #[tokio::test] | |
| 34 | - | async fn s3_delete_failure_keeps_the_row_queued_for_retry() { | |
| 35 | - | let h = TestHarness::with_storage().await; | |
| 36 | - | let storage = h.storage.clone().expect("with_storage provides a backend"); | |
| 37 | - | let key = "test/orphan-retry.bin"; | |
| 38 | - | ||
| 39 | - | storage.put(key, b"payload".to_vec()); | |
| 40 | - | db::pending_s3_deletions::enqueue_deletions( | |
| 41 | - | &h.db, | |
| 42 | - | &[(key.to_string(), "main".to_string())], | |
| 43 | - | "test_failure_path", | |
| 44 | - | ) | |
| 45 | - | .await | |
| 46 | - | .unwrap(); | |
| 47 | - | assert_eq!(queued_deletions(&h, key).await, 1, "row starts queued"); | |
| 48 | - | ||
| 49 | - | storage | |
| 50 | - | .faults() | |
| 51 | - | .fail_always("delete_object", storage_unavailable); | |
| 52 | - | let deleted = h.drain_s3_deletions().await; | |
| 53 | - | ||
| 54 | - | assert_eq!(deleted, 0, "a failing backend deletes nothing"); | |
| 55 | - | assert_eq!( | |
| 56 | - | queued_deletions(&h, key).await, | |
| 57 | - | 1, | |
| 58 | - | "the row must survive a failed delete, dropping it would orphan the object" | |
| 59 | - | ); | |
| 60 | - | assert!( | |
| 61 | - | storage.object_exists(key).await.unwrap(), | |
| 62 | - | "the object is still there, which is why the row must be" | |
| 63 | - | ); | |
| 64 | - | assert_eq!( | |
| 65 | - | storage.faults().calls("delete_object"), | |
| 66 | - | 1, | |
| 67 | - | "the drain attempted the delete exactly once" | |
| 68 | - | ); | |
| 69 | - | } | |
| 70 | - | ||
| 71 | - | /// The point of keeping the row: a later drain finishes the job. This is the | |
| 72 | - | /// whole contract of the durable queue and nothing asserted it before. | |
| 73 | - | #[tokio::test] | |
| 74 | - | async fn s3_delete_queue_recovers_when_the_backend_comes_back() { | |
| 75 | - | let h = TestHarness::with_storage().await; | |
| 76 | - | let storage = h.storage.clone().expect("with_storage provides a backend"); | |
| 77 | - | let key = "test/orphan-recovers.bin"; | |
| 78 | - | ||
| 79 | - | storage.put(key, b"payload".to_vec()); | |
| 80 | - | db::pending_s3_deletions::enqueue_deletions( | |
| 81 | - | &h.db, | |
| 82 | - | &[(key.to_string(), "main".to_string())], | |
| 83 | - | "test_failure_path", | |
| 84 | - | ) | |
| 85 | - | .await | |
| 86 | - | .unwrap(); | |
| 87 | - | ||
| 88 | - | // Down for the first attempt, up for the second. | |
| 89 | - | storage | |
| 90 | - | .faults() | |
| 91 | - | .fail_until("delete_object", 2, storage_unavailable); | |
| 92 | - | ||
| 93 | - | assert_eq!(h.drain_s3_deletions().await, 0, "first drain fails"); | |
| 94 | - | assert_eq!(queued_deletions(&h, key).await, 1, "still queued"); | |
| 95 | - | ||
| 96 | - | assert_eq!(h.drain_s3_deletions().await, 1, "second drain succeeds"); | |
| 97 | - | assert_eq!( | |
| 98 | - | queued_deletions(&h, key).await, | |
| 99 | - | 0, | |
| 100 | - | "a completed delete is dequeued" | |
| 101 | - | ); | |
| 102 | - | assert!( | |
| 103 | - | !storage.object_exists(key).await.unwrap(), | |
| 104 | - | "the object is gone" | |
| 105 | - | ); | |
| 106 | - | } | |
| 107 | - | ||
| 108 | 23 | // Checkout compensation when Stripe is down | |
| 109 | 24 | ||
| 110 | 25 | /// Create a creator with Stripe connected and a published paid item, logged in | |
| @@ -400,312 +315,6 @@ | |||
| 400 | 315 | ); | |
| 401 | 316 | } | |
| 402 | 317 | ||
| 403 | - | // The scan-job retry budget | |
| 404 | - | ||
| 405 | - | /// Set up a trusted creator with an audio item, presign an upload, put the | |
| 406 | - | /// bytes, and confirm it, leaving exactly one queued scan job. Returns the item | |
| 407 | - | /// id and the staging key the job will try to download. | |
| 408 | - | async fn queue_one_scan_job(h: &mut TestHarness) -> (String, String) { | |
| 409 | - | let setup = h.create_creator_with_item("fpscan", "audio", 0).await; | |
| 410 | - | h.trust_user(setup.user_id).await; | |
| 411 | - | h.grant_tier(setup.user_id, "small_files").await; | |
| 412 | - | ||
| 413 | - | let body = serde_json::json!({ | |
| 414 | - | "item_id": setup.item_id, | |
| 415 | - | "file_type": "audio", | |
| 416 | - | "file_name": "held.mp3", | |
| 417 | - | "content_type": "audio/mpeg", | |
| 418 | - | }); | |
| 419 | - | let resp = h | |
| 420 | - | .client | |
| 421 | - | .post_json("/api/upload/presign", &body.to_string()) | |
| 422 | - | .await; | |
| 423 | - | assert_eq!(resp.status, 200, "presign failed: {}", resp.text); | |
| 424 | - | let s3_key = resp.json::<Value>()["s3_key"] | |
| 425 | - | .as_str() | |
| 426 | - | .expect("presign returns s3_key") | |
| 427 | - | .to_string(); | |
| 428 | - | ||
| 429 | - | let mut mp3 = b"ID3".to_vec(); | |
| 430 | - | mp3.extend_from_slice(&[0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]); | |
| 431 | - | mp3.extend_from_slice(&[0u8; 100]); | |
| 432 | - | h.storage.as_ref().unwrap().put(&s3_key, mp3); | |
| 433 | - | ||
| 434 | - | let body = serde_json::json!({ | |
| 435 | - | "item_id": setup.item_id, | |
| 436 | - | "file_type": "audio", | |
| 437 | - | "s3_key": s3_key, | |
| 438 | - | }); | |
| 439 | - | let resp = h | |
| 440 | - | .client | |
| 441 | - | .post_json("/api/upload/confirm", &body.to_string()) | |
| 442 | - | .await; | |
| 443 | - | assert_eq!(resp.status, 200, "confirm failed: {}", resp.text); | |
| 444 | - | ||
| 445 | - | (setup.item_id, s3_key) | |
| 446 | - | } | |
| 447 | - | ||
| 448 | - | async fn job_row(h: &TestHarness, item_id: &str) -> (String, i32, Option<String>) { | |
| 449 | - | sqlx::query_as("SELECT status, attempts, last_error FROM scan_jobs WHERE target_id = $1::uuid") | |
| 450 | - | .bind(item_id) | |
| 451 | - | .fetch_one(&h.db) | |
| 452 | - | .await | |
| 453 | - | .unwrap() | |
| 454 | - | } | |
| 455 | - | ||
| 456 | - | /// A scan whose download fails must record the failure and park the entity at | |
| 457 | - | /// `held_for_review`. Leaving it at `scanning` is the production regression the | |
| 458 | - | /// reset in `process_job` exists to prevent: the file is invisible to the buyer | |
| 459 | - | /// and invisible to the admin queue, so nothing ever resolves it. | |
| 460 | - | #[tokio::test] | |
| 461 | - | async fn scan_download_failure_marks_the_job_failed_and_holds_the_entity() { | |
| 462 | - | let mut h = TestHarness::with_storage_and_scanner().await; | |
| 463 | - | let (item_id, _key) = queue_one_scan_job(&mut h).await; | |
| 464 | - | let storage = h.storage.clone().expect("scanner harness provides storage"); | |
| 465 | - | ||
| 466 | - | // Both scanner read paths (`download_object_buf_capped` for small files, | |
| 467 | - | // `download_stream` for spooled ones) bottom out in `download_stream`, so | |
| 468 | - | // one rule covers the branch either size takes. | |
| 469 | - | storage | |
| 470 | - | .faults() | |
| 471 | - | .fail_always("download_stream", storage_unavailable); | |
| 472 | - | ||
| 473 | - | let err = h | |
| 474 | - | .try_process_one_scan_job() | |
| 475 | - | .await | |
| 476 | - | .expect_err("a failing download must surface as a job error"); | |
| 477 | - | ||
| 478 | - | let (status, attempts, last_error) = job_row(&h, &item_id).await; | |
| 479 | - | assert_eq!(status, "failed", "the job records its own failure"); | |
| 480 | - | assert_eq!(attempts, 1, "the claim consumed exactly one attempt"); | |
| 481 | - | assert!( | |
| 482 | - | last_error.is_some_and(|e| !e.is_empty()), | |
| 483 | - | "last_error is what an admin has to work from" | |
| 484 | - | ); | |
| 485 | - | ||
| 486 | - | let scan_status: String = | |
| 487 | - | sqlx::query_scalar("SELECT scan_status FROM items WHERE id = $1::uuid") | |
| 488 | - | .bind(&item_id) | |
| 489 | - | .fetch_one(&h.db) | |
| 490 | - | .await | |
| 491 | - | .unwrap(); | |
| 492 | - | assert_eq!( | |
| 493 | - | scan_status, "held_for_review", | |
| 494 | - | "a failed scan must not leave the entity stuck at 'scanning'" | |
| 495 | - | ); | |
| 496 | - | assert!( | |
| 497 | - | err.contains("S3") || err.contains("torage"), | |
| 498 | - | "the error should name the failing dependency, got: {err}" | |
| 499 | - | ); | |
| 500 | - | } | |
| 501 | - | ||
| 502 | - | /// A worker that dies mid-scan leaves its row `running` forever; `reap_stuck` is | |
| 503 | - | /// what returns it to the queue. Below the attempt ceiling that is a requeue, | |
| 504 | - | /// and the retry then succeeds once storage is back. Nothing asserted the | |
| 505 | - | /// recovery half before, which is the half the budget exists for. | |
| 506 | - | #[tokio::test] | |
| 507 | - | async fn a_reaped_scan_job_is_requeued_and_succeeds_when_storage_recovers() { | |
| 508 | - | let mut h = TestHarness::with_storage_and_scanner().await; | |
| 509 | - | let (item_id, _key) = queue_one_scan_job(&mut h).await; | |
| 510 | - | let storage = h.storage.clone().expect("scanner harness provides storage"); | |
| 511 | - | ||
| 512 | - | // Claim the job the way a worker would, then abandon it: no mark_done, no | |
| 513 | - | // mark_failed, exactly what a killed process leaves behind. | |
| 514 | - | let job = db::scan_jobs::claim_next(&h.db) | |
| 515 | - | .await | |
| 516 | - | .unwrap() | |
| 517 | - | .expect("the confirm queued a job"); | |
| 518 | - | sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1") | |
| 519 | - | .bind(job.id) | |
| 520 | - | .execute(&h.db) | |
| 521 | - | .await | |
| 522 | - | .unwrap(); | |
| 523 | - | ||
| 524 | - | let reaped = db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap(); | |
| 525 | - | assert_eq!(reaped, 1, "the stale heartbeat is what the reaper keys on"); | |
| 526 | - | ||
| 527 | - | let (status, attempts, _) = job_row(&h, &item_id).await; | |
| 528 | - | assert_eq!( | |
| 529 | - | status, "queued", | |
| 530 | - | "below the ceiling a reaped job goes back to the queue, not to failed" | |
| 531 | - | ); | |
| 532 | - | assert_eq!(attempts, 1, "the abandoned attempt is still spent"); | |
| 533 | - | ||
| 534 | - | // Storage is healthy again; the retry must complete the job. | |
| 535 | - | assert!( | |
| 536 | - | storage.faults().calls("download_stream") == 0, | |
| 537 | - | "no fault installed, the first attempt never reached the backend" | |
| 538 | - | ); | |
| 539 | - | h.drain_scan_jobs().await; | |
| 540 | - | ||
| 541 | - | let (status, attempts, _) = job_row(&h, &item_id).await; | |
| 542 | - | assert_eq!(status, "done", "the retry completes the job"); | |
| 543 | - | assert_eq!(attempts, 2, "the retry consumed a second attempt"); | |
| 544 | - | } | |
| 545 | - | ||
| 546 | - | /// The ceiling is what stops a job that reliably kills its worker from being | |
| 547 | - | /// re-attempted forever. At `MAX_SCAN_ATTEMPTS` the reaper retires the row to | |
| 548 | - | /// `failed` rather than requeueing it, and `claim_next` will not hand it out | |
| 549 | - | /// again. | |
| 550 | - | #[tokio::test] | |
| 551 | - | async fn a_scan_job_at_its_attempt_ceiling_is_retired_not_requeued() { | |
| 552 | - | let mut h = TestHarness::with_storage_and_scanner().await; | |
| 553 | - | let (item_id, _key) = queue_one_scan_job(&mut h).await; | |
| 554 | - | ||
| 555 | - | // Spend the budget down to its last attempt, then claim, which takes it. | |
| 556 | - | sqlx::query("UPDATE scan_jobs SET attempts = $1 WHERE target_id = $2::uuid") | |
| 557 | - | .bind(db::scan_jobs::MAX_SCAN_ATTEMPTS - 1) | |
| 558 | - | .bind(&item_id) | |
| 559 | - | .execute(&h.db) | |
| 560 | - | .await | |
| 561 | - | .unwrap(); | |
| 562 | - | let job = db::scan_jobs::claim_next(&h.db) | |
| 563 | - | .await | |
| 564 | - | .unwrap() | |
| 565 | - | .expect("a job one under the ceiling is still claimable"); | |
| 566 | - | assert_eq!(job.attempts, db::scan_jobs::MAX_SCAN_ATTEMPTS); | |
| 567 | - | ||
| 568 | - | sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1") | |
| 569 | - | .bind(job.id) | |
| 570 | - | .execute(&h.db) | |
| 571 | - | .await | |
| 572 | - | .unwrap(); | |
| 573 | - | assert_eq!(db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap(), 1); | |
| 574 | - | ||
| 575 | - | let (status, _, last_error) = job_row(&h, &item_id).await; | |
| 576 | - | assert_eq!( | |
| 577 | - | status, "failed", | |
| 578 | - | "at the ceiling the reaper retires the job instead of requeueing it" | |
| 579 | - | ); | |
| 580 | - | assert!( | |
| 581 | - | last_error.is_some_and(|e| e.contains("max scan attempts")), | |
| 582 | - | "the retirement reason must be legible to an admin" | |
| 583 | - | ); | |
| 584 | - | ||
| 585 | - | assert!( | |
| 586 | - | db::scan_jobs::claim_next(&h.db).await.unwrap().is_none(), | |
| 587 | - | "a retired job must never be claimed again" | |
| 588 | - | ); | |
| 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 | - | } | |
| 708 | - | ||
| 709 | 318 | // The pending-refund crash window (PAY-S1) | |
| 710 | 319 | ||
| 711 | 320 | /// A refund's claim is deliberately not the same thing as its completion. The |
| @@ -1,0 +1,208 @@ | |||
| 1 | + | //! Negative paths: what the server does when the scanner cannot fetch bytes. | |
| 2 | + | //! | |
| 3 | + | //! Split out of the former `failure_paths.rs` on 2026-08-05, which reached 868 | |
| 4 | + | //! lines and tripped the oversized-module ratchet in `tests/test_hygiene.rs`. | |
| 5 | + | //! The split is by the dependency that fails, which is also how you look these | |
| 6 | + | //! up: the scan-job retry budget and its parking behaviour. | |
| 7 | + | //! | |
| 8 | + | //! These tests exist because the mocks used to be infallible, so the retry and | |
| 9 | + | //! compensation machinery the server carries had no test that could reach it. | |
| 10 | + | //! Retry logic no test can enter is worse than none, because it reads as | |
| 11 | + | //! handled. Each test installs a failure policy on a mock (see | |
| 12 | + | //! `harness::faults`) and asserts the compensating behaviour, not just that the | |
| 13 | + | //! request failed. | |
| 14 | + | //! | |
| 15 | + | //! Rationale: wiki `testing-posture`, the "absent oracle" section. | |
| 16 | + | ||
| 17 | + | use crate::harness::TestHarness; | |
| 18 | + | use crate::harness::faults::storage_unavailable; | |
| 19 | + | use makenotwork::db; | |
| 20 | + | use serde_json::Value; | |
| 21 | + | ||
| 22 | + | // The scan-job retry budget | |
| 23 | + | ||
| 24 | + | /// Set up a trusted creator with an audio item, presign an upload, put the | |
| 25 | + | /// bytes, and confirm it, leaving exactly one queued scan job. Returns the item | |
| 26 | + | /// id and the staging key the job will try to download. | |
| 27 | + | async fn queue_one_scan_job(h: &mut TestHarness) -> (String, String) { | |
| 28 | + | let setup = h.create_creator_with_item("fpscan", "audio", 0).await; | |
| 29 | + | h.trust_user(setup.user_id).await; | |
| 30 | + | h.grant_tier(setup.user_id, "small_files").await; | |
| 31 | + | ||
| 32 | + | let body = serde_json::json!({ | |
| 33 | + | "item_id": setup.item_id, | |
| 34 | + | "file_type": "audio", | |
| 35 | + | "file_name": "held.mp3", | |
| 36 | + | "content_type": "audio/mpeg", | |
| 37 | + | }); | |
| 38 | + | let resp = h | |
| 39 | + | .client | |
| 40 | + | .post_json("/api/upload/presign", &body.to_string()) | |
| 41 | + | .await; | |
| 42 | + | assert_eq!(resp.status, 200, "presign failed: {}", resp.text); | |
| 43 | + | let s3_key = resp.json::<Value>()["s3_key"] | |
| 44 | + | .as_str() | |
| 45 | + | .expect("presign returns s3_key") | |
| 46 | + | .to_string(); | |
| 47 | + | ||
| 48 | + | let mut mp3 = b"ID3".to_vec(); | |
| 49 | + | mp3.extend_from_slice(&[0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]); | |
| 50 | + | mp3.extend_from_slice(&[0u8; 100]); | |
| 51 | + | h.storage.as_ref().unwrap().put(&s3_key, mp3); | |
| 52 | + | ||
| 53 | + | let body = serde_json::json!({ | |
| 54 | + | "item_id": setup.item_id, | |
| 55 | + | "file_type": "audio", | |
| 56 | + | "s3_key": s3_key, | |
| 57 | + | }); | |
| 58 | + | let resp = h | |
| 59 | + | .client | |
| 60 | + | .post_json("/api/upload/confirm", &body.to_string()) | |
| 61 | + | .await; | |
| 62 | + | assert_eq!(resp.status, 200, "confirm failed: {}", resp.text); | |
| 63 | + | ||
| 64 | + | (setup.item_id, s3_key) | |
| 65 | + | } | |
| 66 | + | ||
| 67 | + | async fn job_row(h: &TestHarness, item_id: &str) -> (String, i32, Option<String>) { | |
| 68 | + | sqlx::query_as("SELECT status, attempts, last_error FROM scan_jobs WHERE target_id = $1::uuid") | |
| 69 | + | .bind(item_id) | |
| 70 | + | .fetch_one(&h.db) | |
| 71 | + | .await | |
| 72 | + | .unwrap() | |
| 73 | + | } | |
| 74 | + | ||
| 75 | + | /// A scan whose download fails must record the failure and park the entity at | |
| 76 | + | /// `held_for_review`. Leaving it at `scanning` is the production regression the | |
| 77 | + | /// reset in `process_job` exists to prevent: the file is invisible to the buyer | |
| 78 | + | /// and invisible to the admin queue, so nothing ever resolves it. | |
| 79 | + | #[tokio::test] | |
| 80 | + | async fn scan_download_failure_marks_the_job_failed_and_holds_the_entity() { | |
| 81 | + | let mut h = TestHarness::with_storage_and_scanner().await; | |
| 82 | + | let (item_id, _key) = queue_one_scan_job(&mut h).await; | |
| 83 | + | let storage = h.storage.clone().expect("scanner harness provides storage"); | |
| 84 | + | ||
| 85 | + | // Both scanner read paths (`download_object_buf_capped` for small files, | |
| 86 | + | // `download_stream` for spooled ones) bottom out in `download_stream`, so | |
| 87 | + | // one rule covers the branch either size takes. | |
| 88 | + | storage | |
| 89 | + | .faults() | |
| 90 | + | .fail_always("download_stream", storage_unavailable); | |
| 91 | + | ||
| 92 | + | let err = h | |
| 93 | + | .try_process_one_scan_job() | |
| 94 | + | .await | |
| 95 | + | .expect_err("a failing download must surface as a job error"); | |
| 96 | + | ||
| 97 | + | let (status, attempts, last_error) = job_row(&h, &item_id).await; | |
| 98 | + | assert_eq!(status, "failed", "the job records its own failure"); | |
| 99 | + | assert_eq!(attempts, 1, "the claim consumed exactly one attempt"); | |
| 100 | + | assert!( | |
| 101 | + | last_error.is_some_and(|e| !e.is_empty()), | |
| 102 | + | "last_error is what an admin has to work from" | |
| 103 | + | ); | |
| 104 | + | ||
| 105 | + | let scan_status: String = | |
| 106 | + | sqlx::query_scalar("SELECT scan_status FROM items WHERE id = $1::uuid") | |
| 107 | + | .bind(&item_id) | |
| 108 | + | .fetch_one(&h.db) | |
| 109 | + | .await | |
| 110 | + | .unwrap(); | |
| 111 | + | assert_eq!( | |
| 112 | + | scan_status, "held_for_review", | |
| 113 | + | "a failed scan must not leave the entity stuck at 'scanning'" | |
| 114 | + | ); | |
| 115 | + | assert!( | |
| 116 | + | err.contains("S3") || err.contains("torage"), | |
| 117 | + | "the error should name the failing dependency, got: {err}" | |
| 118 | + | ); | |
| 119 | + | } | |
| 120 | + | ||
| 121 | + | /// A worker that dies mid-scan leaves its row `running` forever; `reap_stuck` is | |
| 122 | + | /// what returns it to the queue. Below the attempt ceiling that is a requeue, | |
| 123 | + | /// and the retry then succeeds once storage is back. Nothing asserted the | |
| 124 | + | /// recovery half before, which is the half the budget exists for. | |
| 125 | + | #[tokio::test] | |
| 126 | + | async fn a_reaped_scan_job_is_requeued_and_succeeds_when_storage_recovers() { | |
| 127 | + | let mut h = TestHarness::with_storage_and_scanner().await; | |
| 128 | + | let (item_id, _key) = queue_one_scan_job(&mut h).await; | |
| 129 | + | let storage = h.storage.clone().expect("scanner harness provides storage"); | |
| 130 | + | ||
| 131 | + | // Claim the job the way a worker would, then abandon it: no mark_done, no | |
| 132 | + | // mark_failed, exactly what a killed process leaves behind. | |
| 133 | + | let job = db::scan_jobs::claim_next(&h.db) | |
| 134 | + | .await | |
| 135 | + | .unwrap() | |
| 136 | + | .expect("the confirm queued a job"); | |
| 137 | + | sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1") | |
| 138 | + | .bind(job.id) | |
| 139 | + | .execute(&h.db) | |
| 140 | + | .await | |
| 141 | + | .unwrap(); | |
| 142 | + | ||
| 143 | + | let reaped = db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap(); | |
| 144 | + | assert_eq!(reaped, 1, "the stale heartbeat is what the reaper keys on"); | |
| 145 | + | ||
| 146 | + | let (status, attempts, _) = job_row(&h, &item_id).await; | |
| 147 | + | assert_eq!( | |
| 148 | + | status, "queued", | |
| 149 | + | "below the ceiling a reaped job goes back to the queue, not to failed" | |
| 150 | + | ); | |
| 151 | + | assert_eq!(attempts, 1, "the abandoned attempt is still spent"); | |
| 152 | + | ||
| 153 | + | // Storage is healthy again; the retry must complete the job. | |
| 154 | + | assert!( | |
| 155 | + | storage.faults().calls("download_stream") == 0, | |
| 156 | + | "no fault installed, the first attempt never reached the backend" | |
| 157 | + | ); | |
| 158 | + | h.drain_scan_jobs().await; | |
| 159 | + | ||
| 160 | + | let (status, attempts, _) = job_row(&h, &item_id).await; | |
| 161 | + | assert_eq!(status, "done", "the retry completes the job"); | |
| 162 | + | assert_eq!(attempts, 2, "the retry consumed a second attempt"); | |
| 163 | + | } | |
| 164 | + | ||
| 165 | + | /// The ceiling is what stops a job that reliably kills its worker from being | |
| 166 | + | /// re-attempted forever. At `MAX_SCAN_ATTEMPTS` the reaper retires the row to | |
| 167 | + | /// `failed` rather than requeueing it, and `claim_next` will not hand it out | |
| 168 | + | /// again. | |
| 169 | + | #[tokio::test] | |
| 170 | + | async fn a_scan_job_at_its_attempt_ceiling_is_retired_not_requeued() { | |
| 171 | + | let mut h = TestHarness::with_storage_and_scanner().await; | |
| 172 | + | let (item_id, _key) = queue_one_scan_job(&mut h).await; | |
| 173 | + | ||
| 174 | + | // Spend the budget down to its last attempt, then claim, which takes it. | |
| 175 | + | sqlx::query("UPDATE scan_jobs SET attempts = $1 WHERE target_id = $2::uuid") | |
| 176 | + | .bind(db::scan_jobs::MAX_SCAN_ATTEMPTS - 1) | |
| 177 | + | .bind(&item_id) | |
| 178 | + | .execute(&h.db) | |
| 179 | + | .await | |
| 180 | + | .unwrap(); | |
| 181 | + | let job = db::scan_jobs::claim_next(&h.db) | |
| 182 | + | .await | |
| 183 | + | .unwrap() | |
| 184 | + | .expect("a job one under the ceiling is still claimable"); | |
| 185 | + | assert_eq!(job.attempts, db::scan_jobs::MAX_SCAN_ATTEMPTS); | |
| 186 | + | ||
| 187 | + | sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1") | |
| 188 | + | .bind(job.id) | |
| 189 | + | .execute(&h.db) | |
| 190 | + | .await | |
| 191 | + | .unwrap(); | |
| 192 | + | assert_eq!(db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap(), 1); | |
| 193 | + | ||
| 194 | + | let (status, _, last_error) = job_row(&h, &item_id).await; | |
| 195 | + | assert_eq!( | |
| 196 | + | status, "failed", | |
| 197 | + | "at the ceiling the reaper retires the job instead of requeueing it" | |
| 198 | + | ); | |
| 199 | + | assert!( | |
| 200 | + | last_error.is_some_and(|e| e.contains("max scan attempts")), | |
| 201 | + | "the retirement reason must be legible to an admin" | |
| 202 | + | ); | |
| 203 | + | ||
| 204 | + | assert!( | |
| 205 | + | db::scan_jobs::claim_next(&h.db).await.unwrap().is_none(), | |
| 206 | + | "a retired job must never be claimed again" | |
| 207 | + | ); | |
| 208 | + | } |
| @@ -1,0 +1,227 @@ | |||
| 1 | + | //! Negative paths: what the server does when object storage fails. | |
| 2 | + | //! | |
| 3 | + | //! Split out of the former `failure_paths.rs` on 2026-08-05, which reached 868 | |
| 4 | + | //! lines and tripped the oversized-module ratchet in `tests/test_hygiene.rs`. | |
| 5 | + | //! The split is by the dependency that fails, which is also how you look these | |
| 6 | + | //! up: anything about S3 being down lands here. | |
| 7 | + | //! | |
| 8 | + | //! These tests exist because the mocks used to be infallible, so the retry and | |
| 9 | + | //! compensation machinery the server carries had no test that could reach it. | |
| 10 | + | //! Retry logic no test can enter is worse than none, because it reads as | |
| 11 | + | //! handled. Each test installs a failure policy on a mock (see | |
| 12 | + | //! `harness::faults`) and asserts the compensating behaviour, not just that the | |
| 13 | + | //! request failed. | |
| 14 | + | //! | |
| 15 | + | //! Rationale: wiki `testing-posture`, the "absent oracle" section. | |
| 16 | + | ||
| 17 | + | use crate::harness::TestHarness; | |
| 18 | + | use crate::harness::faults::storage_unavailable; | |
| 19 | + | use makenotwork::db; | |
| 20 | + | use makenotwork::storage::StorageBackend; | |
| 21 | + | ||
| 22 | + | // The durable S3 deletion queue | |
| 23 | + | ||
| 24 | + | /// Count rows still queued for deletion of `key`. | |
| 25 | + | async fn queued_deletions(h: &TestHarness, key: &str) -> i64 { | |
| 26 | + | sqlx::query_scalar("SELECT COUNT(*) FROM pending_s3_deletions WHERE s3_key = $1") | |
| 27 | + | .bind(key) | |
| 28 | + | .fetch_one(&h.db) | |
| 29 | + | .await | |
| 30 | + | .unwrap() | |
| 31 | + | } | |
| 32 | + | ||
| 33 | + | /// A delete that fails must leave the row queued. Dequeuing it would orphan the | |
| 34 | + | /// S3 object with no durable record, which is the leak the queue exists to | |
| 35 | + | /// prevent. | |
| 36 | + | #[tokio::test] | |
| 37 | + | async fn s3_delete_failure_keeps_the_row_queued_for_retry() { | |
| 38 | + | let h = TestHarness::with_storage().await; | |
| 39 | + | let storage = h.storage.clone().expect("with_storage provides a backend"); | |
| 40 | + | let key = "test/orphan-retry.bin"; | |
| 41 | + | ||
| 42 | + | storage.put(key, b"payload".to_vec()); | |
| 43 | + | db::pending_s3_deletions::enqueue_deletions( | |
| 44 | + | &h.db, | |
| 45 | + | &[(key.to_string(), "main".to_string())], | |
| 46 | + | "test_failure_path", | |
| 47 | + | ) | |
| 48 | + | .await | |
| 49 | + | .unwrap(); | |
| 50 | + | assert_eq!(queued_deletions(&h, key).await, 1, "row starts queued"); | |
| 51 | + | ||
| 52 | + | storage | |
| 53 | + | .faults() | |
| 54 | + | .fail_always("delete_object", storage_unavailable); | |
| 55 | + | let deleted = h.drain_s3_deletions().await; | |
| 56 | + | ||
| 57 | + | assert_eq!(deleted, 0, "a failing backend deletes nothing"); | |
| 58 | + | assert_eq!( | |
| 59 | + | queued_deletions(&h, key).await, | |
| 60 | + | 1, | |
| 61 | + | "the row must survive a failed delete, dropping it would orphan the object" | |
| 62 | + | ); | |
| 63 | + | assert!( | |
| 64 | + | storage.object_exists(key).await.unwrap(), | |
| 65 | + | "the object is still there, which is why the row must be" | |
| 66 | + | ); | |
| 67 | + | assert_eq!( | |
| 68 | + | storage.faults().calls("delete_object"), | |
| 69 | + | 1, | |
| 70 | + | "the drain attempted the delete exactly once" | |
| 71 | + | ); | |
| 72 | + | } | |
| 73 | + | ||
| 74 | + | /// The point of keeping the row: a later drain finishes the job. This is the | |
| 75 | + | /// whole contract of the durable queue and nothing asserted it before. | |
| 76 | + | #[tokio::test] | |
| 77 | + | async fn s3_delete_queue_recovers_when_the_backend_comes_back() { | |
| 78 | + | let h = TestHarness::with_storage().await; | |
| 79 | + | let storage = h.storage.clone().expect("with_storage provides a backend"); | |
| 80 | + | let key = "test/orphan-recovers.bin"; | |
| 81 | + | ||
| 82 | + | storage.put(key, b"payload".to_vec()); | |
| 83 | + | db::pending_s3_deletions::enqueue_deletions( | |
| 84 | + | &h.db, | |
| 85 | + | &[(key.to_string(), "main".to_string())], | |
| 86 | + | "test_failure_path", | |
| 87 | + | ) | |
| 88 | + | .await | |
| 89 | + | .unwrap(); | |
| 90 | + | ||
| 91 | + | // Down for the first attempt, up for the second. | |
| 92 | + | storage | |
| 93 | + | .faults() | |
| 94 | + | .fail_until("delete_object", 2, storage_unavailable); | |
| 95 | + | ||
| 96 | + | assert_eq!(h.drain_s3_deletions().await, 0, "first drain fails"); | |
| 97 | + | assert_eq!(queued_deletions(&h, key).await, 1, "still queued"); | |
| 98 | + | ||
| 99 | + | assert_eq!(h.drain_s3_deletions().await, 1, "second drain succeeds"); | |
| 100 | + | assert_eq!( | |
| 101 | + | queued_deletions(&h, key).await, | |
| 102 | + | 0, | |
| 103 | + | "a completed delete is dequeued" | |
| 104 | + | ); | |
| 105 | + | assert!( | |
| 106 | + | !storage.object_exists(key).await.unwrap(), | |
| 107 | + | "the object is gone" | |
| 108 | + | ); | |
| 109 | + | } | |
| 110 | + | ||
| 111 | + | // The orphaned-upload reaper | |
| 112 | + | ||
| 113 | + | /// Insert a pending upload that is already old enough for the reaper, with the | |
| 114 | + | /// object present in storage. Returns the key. | |
| 115 | + | async fn stale_pending_upload(h: &TestHarness, user_id: db::UserId, key: &str) -> String { | |
| 116 | + | h.storage.as_ref().unwrap().put(key, b"orphan".to_vec()); | |
| 117 | + | sqlx::query( | |
| 118 | + | "INSERT INTO pending_uploads (user_id, s3_key, bucket, created_at) | |
| 119 | + | VALUES ($1, $2, 'main', NOW() - INTERVAL '48 hours')", | |
| 120 | + | ) | |
| 121 | + | .bind(user_id) | |
| 122 | + | .bind(key) | |
| 123 | + | .execute(&h.db) | |
| 124 | + | .await | |
| 125 | + | .unwrap(); | |
| 126 | + | key.to_string() | |
| 127 | + | } | |
| 128 | + | ||
| 129 | + | async fn pending_upload_rows(h: &TestHarness, key: &str) -> i64 { | |
| 130 | + | sqlx::query_scalar("SELECT COUNT(*) FROM pending_uploads WHERE s3_key = $1") | |
| 131 | + | .bind(key) | |
| 132 | + | .fetch_one(&h.db) | |
| 133 | + | .await | |
| 134 | + | .unwrap() | |
| 135 | + | } | |
| 136 | + | ||
| 137 | + | /// The happy path, asserted here so the failure path below is a contrast rather | |
| 138 | + | /// than the only thing observed: a reaped orphan is deleted, its tracking row is | |
| 139 | + | /// cleared, and nothing is handed to the durable queue. | |
| 140 | + | #[tokio::test] | |
| 141 | + | async fn the_reaper_deletes_an_orphan_and_clears_its_row() { | |
| 142 | + | let mut h = TestHarness::with_storage().await; | |
| 143 | + | let user_id = h.signup("reap1", "reap1@test.com", "pass1234").await; | |
| 144 | + | let key = stale_pending_upload(&h, user_id, "staging/reaped.bin").await; | |
| 145 | + | let storage = h.storage.clone().unwrap(); | |
| 146 | + | ||
| 147 | + | h.run_orphan_upload_reaper().await; | |
| 148 | + | ||
| 149 | + | assert!( | |
| 150 | + | !storage.object_exists(&key).await.unwrap(), | |
| 151 | + | "the orphan object is deleted" | |
| 152 | + | ); | |
| 153 | + | assert_eq!( | |
| 154 | + | pending_upload_rows(&h, &key).await, | |
| 155 | + | 0, | |
| 156 | + | "tracking row cleared" | |
| 157 | + | ); | |
| 158 | + | assert_eq!( | |
| 159 | + | queued_deletions(&h, &key).await, | |
| 160 | + | 0, | |
| 161 | + | "a successful delete must not also enqueue, that would double-handle the key" | |
| 162 | + | ); | |
| 163 | + | } | |
| 164 | + | ||
| 165 | + | /// A transient S3 failure must hand the key to the durable deletion queue | |
| 166 | + | /// BEFORE the tracking row is cleared. Clearing the row on a transient failure | |
| 167 | + | /// dropped the only record of the object and leaked it permanently (Run #2 | |
| 168 | + | /// Storage SERIOUS). The fix has been in the tree unobserved since; this is the | |
| 169 | + | /// test that enters it. | |
| 170 | + | #[tokio::test] | |
| 171 | + | async fn a_transient_delete_failure_hands_the_orphan_to_the_durable_queue() { | |
| 172 | + | let mut h = TestHarness::with_storage().await; | |
| 173 | + | let user_id = h.signup("reap2", "reap2@test.com", "pass1234").await; | |
| 174 | + | let key = stale_pending_upload(&h, user_id, "staging/handed-off.bin").await; | |
| 175 | + | let storage = h.storage.clone().unwrap(); | |
| 176 | + | ||
| 177 | + | storage | |
| 178 | + | .faults() | |
| 179 | + | .fail_always("delete_object", storage_unavailable); | |
| 180 | + | h.run_orphan_upload_reaper().await; | |
| 181 | + | ||
| 182 | + | assert!( | |
| 183 | + | storage.object_exists(&key).await.unwrap(), | |
| 184 | + | "the delete failed, so the object is still there" | |
| 185 | + | ); | |
| 186 | + | assert_eq!( | |
| 187 | + | queued_deletions(&h, &key).await, | |
| 188 | + | 1, | |
| 189 | + | "the key must be queued for retry; without this the object leaks" | |
| 190 | + | ); | |
| 191 | + | assert_eq!( | |
| 192 | + | pending_upload_rows(&h, &key).await, | |
| 193 | + | 0, | |
| 194 | + | "the tracking row is cleared only because the durable queue now owns the key" | |
| 195 | + | ); | |
| 196 | + | ||
| 197 | + | // The handoff is worth nothing if the queue cannot then finish the job. | |
| 198 | + | storage.faults().clear("delete_object"); | |
| 199 | + | assert_eq!(h.drain_s3_deletions().await, 1, "the retry completes it"); | |
| 200 | + | assert!(!storage.object_exists(&key).await.unwrap(), "object gone"); | |
| 201 | + | } | |
| 202 | + | ||
| 203 | + | /// Aborting orphaned multipart sessions is documented best-effort: it must not | |
| 204 | + | /// block the object delete. A failing abort that stranded the delete would leave | |
| 205 | + | /// the orphan in place every tick forever, and the tracking row with it. | |
| 206 | + | #[tokio::test] | |
| 207 | + | async fn a_failed_multipart_abort_does_not_block_the_orphan_delete() { | |
| 208 | + | let mut h = TestHarness::with_storage().await; | |
| 209 | + | let user_id = h.signup("reap3", "reap3@test.com", "pass1234").await; | |
| 210 | + | let key = stale_pending_upload(&h, user_id, "staging/abort-fails.bin").await; | |
| 211 | + | let storage = h.storage.clone().unwrap(); | |
| 212 | + | ||
| 213 | + | storage | |
| 214 | + | .faults() | |
| 215 | + | .fail_always("list_multipart_uploads_for_key", storage_unavailable); | |
| 216 | + | h.run_orphan_upload_reaper().await; | |
| 217 | + | ||
| 218 | + | assert!( | |
| 219 | + | !storage.object_exists(&key).await.unwrap(), | |
| 220 | + | "a failed abort is best-effort and must not stop the delete" | |
| 221 | + | ); | |
| 222 | + | assert_eq!( | |
| 223 | + | pending_upload_rows(&h, &key).await, | |
| 224 | + | 0, | |
| 225 | + | "and the tracking row is still cleared" | |
| 226 | + | ); | |
| 227 | + | } |