Skip to main content

max / makenotwork

8.0 KB · 209 lines History Blame Raw
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 }
209