Skip to main content

max / makenotwork

Storage: atomic quarantine purge, scan-job retry budget, honest signing chips - purge_cdn_image_rows_by_key wraps its DELETE/UPDATE set in one transaction so a mid-loop failure can't leave a partially-purged malicious reference set. - Scan-job retry budget is now real: claim_next skips jobs at MAX_SCAN_ATTEMPTS and reap_stuck retires a stuck-in-running job to failed once it hits the cap (each claim increments attempts), instead of re-queuing a crash-looping job forever. The reap_stuck doc claimed a budget that did not exist. - Signing layers are presence detectors, not verifiers (they can never emit Fail); reword the macOS/Windows detail chips to 'present (unverified)' so the admin surface doesn't read them as a verified trust verdict. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-01 17:47 UTC
Commit: d2d472c9858d74529ced83424b9d070458b13eb5
Parent: 4192670
4 files changed, +44 insertions, -17 deletions
M server/src/db/scan_jobs.rs +24 -7
@@ -161,18 +161,26 @@ pub async fn enqueue(
161 161 Ok(id)
162 162 }
163 163
164 + /// Maximum times a scan job may be claimed before it is given up as `failed`.
165 + /// Bounds the crash-loop retry budget: a job that reliably wedges the OS process
166 + /// (so it's reaped from `running` rather than reaching `mark_failed`) can only
167 + /// be re-attempted this many times before `reap_stuck` retires it.
168 + pub const MAX_SCAN_ATTEMPTS: i32 = 5;
169 +
164 170 /// Atomically claim the next queued scan job for processing.
165 171 ///
166 172 /// Uses `FOR UPDATE SKIP LOCKED` so multiple workers can drain the queue in
167 173 /// parallel without contention. Sets `status='running'`, increments `attempts`,
168 - /// and stamps `started_at`. Returns `Ok(None)` if the queue is empty.
174 + /// and stamps `started_at`. Skips jobs that have already hit `MAX_SCAN_ATTEMPTS`
175 + /// (those are retired to `failed` by `reap_stuck`). Returns `Ok(None)` if the
176 + /// queue is empty.
169 177 #[tracing::instrument(skip_all)]
170 178 pub async fn claim_next(db: &PgPool) -> Result<Option<ScanJob>, sqlx::Error> {
171 179 let job = sqlx::query_as::<_, ScanJob>(
172 180 r#"
173 181 WITH next AS (
174 182 SELECT id FROM scan_jobs
175 - WHERE status = 'queued'
183 + WHERE status = 'queued' AND attempts < $1
176 184 ORDER BY enqueued_at ASC
177 185 FOR UPDATE SKIP LOCKED
178 186 LIMIT 1
@@ -185,6 +193,7 @@ pub async fn claim_next(db: &PgPool) -> Result<Option<ScanJob>, sqlx::Error> {
185 193 RETURNING *
186 194 "#,
187 195 )
196 + .bind(MAX_SCAN_ATTEMPTS)
188 197 .fetch_optional(db)
189 198 .await?;
190 199
@@ -230,21 +239,29 @@ pub async fn mark_failed(db: &PgPool, job_id: Uuid, err: &str) -> Result<(), sql
230 239
231 240 /// Reset jobs that have been stuck in `running` longer than `max_age_secs`.
232 241 ///
233 - /// Run on worker startup to recover from a previous-process crash mid-scan:
234 - /// the row would otherwise stay `running` forever and never be re-claimed.
235 - /// Increments `attempts` so a perpetually-crashing scan eventually trips the
236 - /// retry budget.
242 + /// Run on worker startup and on a timer to recover from a previous-process crash
243 + /// mid-scan: the row would otherwise stay `running` forever and never be
244 + /// re-claimed. A job whose `attempts` has already reached `MAX_SCAN_ATTEMPTS` is
245 + /// retired to `failed` (each claim increments `attempts`, so a scan that
246 + /// reliably wedges the process trips the budget); otherwise it returns to
247 + /// `queued` for another attempt.
237 248 #[tracing::instrument(skip_all)]
238 249 pub async fn reap_stuck(db: &PgPool, max_age_secs: i64) -> Result<u64, sqlx::Error> {
239 250 let affected = sqlx::query(
240 251 r#"
241 252 UPDATE scan_jobs
242 - SET status = 'queued', started_at = NULL
253 + SET status = CASE WHEN attempts >= $2 THEN 'failed' ELSE 'queued' END,
254 + started_at = NULL,
255 + completed_at = CASE WHEN attempts >= $2 THEN NOW() ELSE completed_at END,
256 + last_error = CASE WHEN attempts >= $2
257 + THEN 'exceeded max scan attempts (reaped from running)'
258 + ELSE last_error END
243 259 WHERE status = 'running'
244 260 AND started_at < NOW() - ($1 || ' seconds')::interval
245 261 "#,
246 262 )
247 263 .bind(max_age_secs.to_string())
264 + .bind(MAX_SCAN_ATTEMPTS)
248 265 .execute(db)
249 266 .await?
250 267 .rows_affected();
M server/src/db/scanning.rs +7 -2
@@ -63,13 +63,17 @@ pub struct HeldVersionRow {
63 63 /// suffix) are registered in `S3_KEY_REFS`, so NULLing both makes the key dead.
64 64 #[tracing::instrument(skip_all)]
65 65 pub async fn purge_cdn_image_rows_by_key(db: &PgPool, s3_key: &str) -> Result<u64, sqlx::Error> {
66 + // Wrap all reference-clearing statements in one transaction: a mid-loop
67 + // failure must not leave a partially-purged malicious reference set (some
68 + // tables still pointing at a quarantined object while others were cleared).
69 + let mut tx = db.begin().await?;
66 70 let mut removed = 0u64;
67 71 for sql in [
68 72 "DELETE FROM item_images WHERE s3_key = $1",
69 73 "DELETE FROM project_images WHERE s3_key = $1",
70 74 "DELETE FROM content_insertions WHERE storage_key = $1",
71 75 ] {
72 - removed += sqlx::query(sql).bind(s3_key).execute(db).await?.rows_affected();
76 + removed += sqlx::query(sql).bind(s3_key).execute(&mut *tx).await?.rows_affected();
73 77 }
74 78 // Item cover: NULL the columns in place, keeping the track. Matches on the
75 79 // exact key in `cover_s3_key` (the cover_image_url suffix is cleared in the
@@ -79,9 +83,10 @@ pub async fn purge_cdn_image_rows_by_key(db: &PgPool, s3_key: &str) -> Result<u6
79 83 cover_file_size_bytes = NULL, updated_at = NOW() WHERE cover_s3_key = $1",
80 84 )
81 85 .bind(s3_key)
82 - .execute(db)
86 + .execute(&mut *tx)
83 87 .await?
84 88 .rows_affected();
89 + tx.commit().await?;
85 90 Ok(removed)
86 91 }
87 92
M server/src/scanning/signing_macos.rs +11 -6
@@ -243,21 +243,25 @@ fn classify(
243 243 format!(" team(s): {}", signed_teams.join(", "))
244 244 };
245 245
246 + // These details describe *presence* of signing/notarization structures, not
247 + // a verified trust chain (this layer never validates the CMS chain and can
248 + // never emit Fail). Word them as "present (unverified)" so the admin surface
249 + // doesn't read them as a green trust verdict.
246 250 match (had_signature, notarization) {
247 251 (true, NotarizationState::Stapled) if !signed_teams.is_empty() => {
248 - pass(format!("Notarized by Apple,{team_str}"))
252 + pass(format!("Notarized by Apple (ticket present, unverified),{team_str}"))
249 253 }
250 254 (true, NotarizationState::Stapled) => {
251 - pass("Notarized by Apple, no team identifier".to_string())
255 + pass("Notarized by Apple (ticket present, unverified), no team identifier".to_string())
252 256 }
253 257 (true, NotarizationState::Malformed) => {
254 - pass(format!("Signed{team_str}; staple present but malformed"))
258 + pass(format!("Signature present{team_str} (unverified); staple malformed"))
255 259 }
256 260 (true, NotarizationState::NotStapled) if !signed_teams.is_empty() => {
257 - pass(format!("Signed by{team_str}, not notarized"))
261 + pass(format!("Signature present by{team_str} (unverified), not notarized"))
258 262 }
259 263 (true, NotarizationState::NotStapled) => {
260 - pass("Signed, no team identifier present".to_string())
264 + pass("Signature present (unverified), no team identifier".to_string())
261 265 }
262 266 (false, _) => pass("Apple binary, no signature".to_string()),
263 267 }
@@ -387,7 +391,8 @@ mod tests {
387 391 let r = classify(true, &["TEAM".to_string()], NotarizationState::Malformed);
388 392 assert_eq!(r.verdict, LayerVerdict::Pass);
389 393 let d = r.detail.unwrap();
390 - assert!(d.contains("staple present but malformed"));
394 + assert!(d.contains("staple malformed"));
395 + assert!(d.contains("unverified"));
391 396 }
392 397
393 398 #[test]
M server/src/scanning/signing_windows.rs +2 -2
@@ -177,8 +177,8 @@ fn extract_subject_cn(cert: &x509_cert::Certificate) -> Option<String> {
177 177
178 178 fn classify(had_signature: bool, signers: &[String]) -> LayerResult {
179 179 match (had_signature, signers.is_empty()) {
180 - (true, false) => pass(format!("Signed by: {}", signers.join(", "))),
181 - (true, true) => pass("Signed, no subject name extractable".to_string()),
180 + (true, false) => pass(format!("Signature present (unverified) by: {}", signers.join(", "))),
181 + (true, true) => pass("Signature present (unverified), no subject name extractable".to_string()),
182 182 (false, _) => pass("PE present, no embedded signature".to_string()),
183 183 }
184 184 }