Skip to main content

max / makenotwork

7.9 KB · 208 lines History Blame Raw
1 //! Negative paths: what the server does when the scanner cannot fetch bytes.
2 //!
3 //! Negative paths are split by the dependency that fails, which is also how you
4 //! look these
5 //! up: the scan-job retry budget and its parking behaviour.
6 //!
7 //! These tests exist because an infallible mock leaves the retry and
8 //! compensation machinery the server carries with no test that can reach it.
9 //! Retry logic no test can enter is worse than none, because it reads as
10 //! handled. Each test installs a failure policy on a mock (see
11 //! `harness::faults`) and asserts the compensating behaviour, not just that the
12 //! request failed.
13 //!
14 //! Rationale: wiki `testing-posture`, the "absent oracle" section.
15
16 use crate::harness::TestHarness;
17 use crate::harness::faults::storage_unavailable;
18 use makenotwork::db;
19 use serde_json::Value;
20
21 // The scan-job retry budget
22
23 /// Set up a trusted creator with an audio item, presign an upload, put the
24 /// bytes, and confirm it, leaving exactly one queued scan job. Returns the item
25 /// id and the staging key the job will try to download.
26 async fn queue_one_scan_job(h: &mut TestHarness) -> (String, String) {
27 let setup = h.create_creator_with_item("fpscan", "audio", 0).await;
28 h.trust_user(setup.user_id).await;
29 h.grant_tier(setup.user_id, "small_files").await;
30
31 let body = serde_json::json!({
32 "item_id": setup.item_id,
33 "file_type": "audio",
34 "file_name": "held.mp3",
35 "content_type": "audio/mpeg",
36 });
37 let resp = h
38 .client
39 .post_json("/api/upload/presign", &body.to_string())
40 .await;
41 assert_eq!(resp.status, 200, "presign failed: {}", resp.text);
42 let s3_key = resp.json::<Value>()["s3_key"]
43 .as_str()
44 .expect("presign returns s3_key")
45 .to_string();
46
47 let mut mp3 = b"ID3".to_vec();
48 mp3.extend_from_slice(&[0x04, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00]);
49 mp3.extend_from_slice(&[0u8; 100]);
50 h.storage.as_ref().unwrap().put(&s3_key, mp3);
51
52 let body = serde_json::json!({
53 "item_id": setup.item_id,
54 "file_type": "audio",
55 "s3_key": s3_key,
56 });
57 let resp = h
58 .client
59 .post_json("/api/upload/confirm", &body.to_string())
60 .await;
61 assert_eq!(resp.status, 200, "confirm failed: {}", resp.text);
62
63 (setup.item_id, s3_key)
64 }
65
66 async fn job_row(h: &TestHarness, item_id: &str) -> (String, i32, Option<String>) {
67 sqlx::query_as("SELECT status, attempts, last_error FROM scan_jobs WHERE target_id = $1::uuid")
68 .bind(item_id)
69 .fetch_one(&h.db)
70 .await
71 .unwrap()
72 }
73
74 /// A scan whose download fails must record the failure and park the entity at
75 /// `held_for_review`. Leaving it at `scanning` is the production regression the
76 /// reset in `process_job` exists to prevent: the file is invisible to the buyer
77 /// and invisible to the admin queue, so nothing ever resolves it.
78 #[tokio::test]
79 async fn scan_download_failure_marks_the_job_failed_and_holds_the_entity() {
80 let mut h = TestHarness::with_storage_and_scanner().await;
81 let (item_id, _key) = queue_one_scan_job(&mut h).await;
82 let storage = h.storage.clone().expect("scanner harness provides storage");
83
84 // Both scanner read paths (`download_object_buf_capped` for small files,
85 // `download_stream` for spooled ones) bottom out in `download_stream`, so
86 // one rule covers the branch either size takes.
87 storage
88 .faults()
89 .fail_always("download_stream", storage_unavailable);
90
91 let err = h
92 .try_process_one_scan_job()
93 .await
94 .expect_err("a failing download must surface as a job error");
95
96 let (status, attempts, last_error) = job_row(&h, &item_id).await;
97 assert_eq!(status, "failed", "the job records its own failure");
98 assert_eq!(attempts, 1, "the claim consumed exactly one attempt");
99 assert!(
100 last_error.is_some_and(|e| !e.is_empty()),
101 "last_error is what an admin has to work from"
102 );
103
104 let scan_status: String =
105 sqlx::query_scalar("SELECT scan_status FROM items WHERE id = $1::uuid")
106 .bind(&item_id)
107 .fetch_one(&h.db)
108 .await
109 .unwrap();
110 assert_eq!(
111 scan_status, "held_for_review",
112 "a failed scan must not leave the entity stuck at 'scanning'"
113 );
114 assert!(
115 err.contains("S3") || err.contains("torage"),
116 "the error should name the failing dependency, got: {err}"
117 );
118 }
119
120 /// A worker that dies mid-scan leaves its row `running` forever; `reap_stuck` is
121 /// what returns it to the queue. Below the attempt ceiling that is a requeue,
122 /// and the retry then succeeds once storage is back. Nothing asserted the
123 /// recovery half before, which is the half the budget exists for.
124 #[tokio::test]
125 async fn a_reaped_scan_job_is_requeued_and_succeeds_when_storage_recovers() {
126 let mut h = TestHarness::with_storage_and_scanner().await;
127 let (item_id, _key) = queue_one_scan_job(&mut h).await;
128 let storage = h.storage.clone().expect("scanner harness provides storage");
129
130 // Claim the job the way a worker would, then abandon it: no mark_done, no
131 // mark_failed, exactly what a killed process leaves behind.
132 let job = db::scan_jobs::claim_next(&h.db)
133 .await
134 .unwrap()
135 .expect("the confirm queued a job");
136 sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1")
137 .bind(job.id)
138 .execute(&h.db)
139 .await
140 .unwrap();
141
142 let reaped = db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap();
143 assert_eq!(reaped, 1, "the stale heartbeat is what the reaper keys on");
144
145 let (status, attempts, _) = job_row(&h, &item_id).await;
146 assert_eq!(
147 status, "queued",
148 "below the ceiling a reaped job goes back to the queue, not to failed"
149 );
150 assert_eq!(attempts, 1, "the abandoned attempt is still spent");
151
152 // Storage is healthy again; the retry must complete the job.
153 assert!(
154 storage.faults().calls("download_stream") == 0,
155 "no fault installed, the first attempt never reached the backend"
156 );
157 h.drain_scan_jobs().await;
158
159 let (status, attempts, _) = job_row(&h, &item_id).await;
160 assert_eq!(status, "done", "the retry completes the job");
161 assert_eq!(attempts, 2, "the retry consumed a second attempt");
162 }
163
164 /// The ceiling is what stops a job that reliably kills its worker from being
165 /// re-attempted forever. At `MAX_SCAN_ATTEMPTS` the reaper retires the row to
166 /// `failed` rather than requeueing it, and `claim_next` will not hand it out
167 /// again.
168 #[tokio::test]
169 async fn a_scan_job_at_its_attempt_ceiling_is_retired_not_requeued() {
170 let mut h = TestHarness::with_storage_and_scanner().await;
171 let (item_id, _key) = queue_one_scan_job(&mut h).await;
172
173 // Spend the budget down to its last attempt, then claim, which takes it.
174 sqlx::query("UPDATE scan_jobs SET attempts = $1 WHERE target_id = $2::uuid")
175 .bind(db::scan_jobs::MAX_SCAN_ATTEMPTS - 1)
176 .bind(&item_id)
177 .execute(&h.db)
178 .await
179 .unwrap();
180 let job = db::scan_jobs::claim_next(&h.db)
181 .await
182 .unwrap()
183 .expect("a job one under the ceiling is still claimable");
184 assert_eq!(job.attempts, db::scan_jobs::MAX_SCAN_ATTEMPTS);
185
186 sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - INTERVAL '1 hour' WHERE id = $1")
187 .bind(job.id)
188 .execute(&h.db)
189 .await
190 .unwrap();
191 assert_eq!(db::scan_jobs::reap_stuck(&h.db, 60).await.unwrap(), 1);
192
193 let (status, _, last_error) = job_row(&h, &item_id).await;
194 assert_eq!(
195 status, "failed",
196 "at the ceiling the reaper retires the job instead of requeueing it"
197 );
198 assert!(
199 last_error.is_some_and(|e| e.contains("max scan attempts")),
200 "the retirement reason must be legible to an admin"
201 );
202
203 assert!(
204 db::scan_jobs::claim_next(&h.db).await.unwrap().is_none(),
205 "a retired job must never be claimed again"
206 );
207 }
208