Skip to main content

max / makenotwork

Reach the scan worker's failure branch from a test The scan-job retry machinery had no test that could enter it. The only entry point, `drain_scan_jobs`, panics on a job error, so the branch that marks a job failed and resets its entity to `held_for_review` was unreachable by construction. Add `try_process_one_scan_job`, which returns the outcome instead, and three tests over the retry budget: a failing download, a reaped job that is requeued and then succeeds, and a job at `MAX_SCAN_ATTEMPTS` that is retired rather than requeued. Both oracles were checked by breaking the code they guard: removing the `HeldForReview` reset fails the first test, and collapsing `reap_stuck`'s attempts ceiling to an unconditional requeue fails the third. Phase 1 of wiki `testing-posture`.
Co-Authored-By
Claude Opus 5 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-08-05 13:54 UTC
Signed with PGP, not checked
Commit: 361c40444cd71598fa781eb872874ec7d1ce3ee3
Parent: 9efc237
2 files changed, +226 insertions, -12 deletions
M server/tests/harness/mod.rs +38 -12
@@ -636,8 +636,43 @@
636 636 /// without spawning a background task, integration tests call this
637 637 /// between upload-confirm and any assertion on `scan_status`.
638 638 pub(crate) async fn drain_scan_jobs(&self) {
639 - let Some(deps) = &self.scan_deps else { return };
640 - let ctx = makenotwork::scanning::worker::WorkerContext {
639 + let Some(ctx) = self.scan_worker_context() else {
640 + return;
641 + };
642 + // Hard cap to avoid an infinite loop if a job re-enqueues itself.
643 + for _ in 0..256 {
644 + match makenotwork::scanning::worker::process_next_for_test(&ctx).await {
645 + Ok(true) => {}
646 + Ok(false) => return,
647 + Err(e) => panic!("scan worker drain failed: {e}"),
648 + }
649 + }
650 + panic!("drain_scan_jobs did not terminate within 256 iterations");
651 + }
652 +
653 + /// Run one scan job and return its outcome instead of panicking on failure.
654 + ///
655 + /// `drain_scan_jobs` treats a failing job as a broken test, which is right
656 + /// for the happy paths but makes the worker's failure branch unobservable:
657 + /// a job whose download fails marks itself `failed` and resets its entity to
658 + /// `HeldForReview`, and no test could reach that while the only entry point
659 + /// panicked. `Ok(true)` ran a job, `Ok(false)` found an empty queue, `Err`
660 + /// carries the message the worker recorded in `last_error`.
661 + #[allow(dead_code)]
662 + pub(crate) async fn try_process_one_scan_job(&self) -> Result<bool, String> {
663 + let Some(ctx) = self.scan_worker_context() else {
664 + return Ok(false);
665 + };
666 + makenotwork::scanning::worker::process_next_for_test(&ctx)
667 + .await
668 + .map_err(|e| e.to_string())
669 + }
670 +
671 + /// The worker context both drain paths run against. `None` when the harness
672 + /// was not built with a scanner.
673 + fn scan_worker_context(&self) -> Option<makenotwork::scanning::worker::WorkerContext> {
674 + let deps = self.scan_deps.as_ref()?;
675 + Some(makenotwork::scanning::worker::WorkerContext {
641 676 db: self.db.clone(),
642 677 s3: deps.s3.clone(),
643 678 pipeline: deps.pipeline.clone(),
@@ -651,16 +686,7 @@
651 686 synckit_s3: Some(deps.s3.clone()),
652 687 // Public bucket shares the same backend; image promotes copy here.
653 688 public_s3: Some(deps.s3.clone()),
654 - };
655 - // Hard cap to avoid an infinite loop if a job re-enqueues itself.
656 - for _ in 0..256 {
657 - match makenotwork::scanning::worker::process_next_for_test(&ctx).await {
658 - Ok(true) => {}
659 - Ok(false) => return,
660 - Err(e) => panic!("scan worker drain failed: {e}"),
661 - }
662 - }
663 - panic!("drain_scan_jobs did not terminate within 256 iterations");
689 + })
664 690 }
665 691
666 692 /// Synchronously perform any queued S3 object deletions (`main` bucket).
M server/tests/workflows/failure_paths.rs +188
@@ -399,3 +399,191 @@
399 399 "an email outage must not touch the purchase"
400 400 );
401 401 }
402 +
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 + }