Skip to main content

max / makenotwork

5.4 KB · 151 lines History Blame Raw
1 //! DB-layer contract tests for the scan-job reaper's liveness heartbeat.
2 //!
3 //! Audit Run 22: `reap_stuck` keyed off claim-time `started_at`, which spans the
4 //! full S3 download + scan, so a slow-but-progressing large scan was reaped and
5 //! double-processed, each re-claim inflating `attempts` until a valid file was
6 //! force-retired to `failed` and the entity stranded at HeldForReview. The
7 //! reaper now keys off `heartbeat_at` (falling back to `started_at`). These pin
8 //! that a fresh beat spares a long-running job, a stale beat is still reaped, the
9 //! NULL fallback works, and `bump_heartbeat` refreshes liveness without
10 //! resurrecting a terminal row.
11
12 use crate::harness::TestHarness;
13 use crate::harness::seed_user;
14 use makenotwork::db::UserId;
15 use makenotwork::db::scan_jobs::{self, ScanTargetKind};
16 use makenotwork::storage::FileType;
17 use uuid::Uuid;
18
19 async fn status_of(h: &TestHarness, job: Uuid) -> String {
20 sqlx::query_scalar::<_, String>("SELECT status FROM scan_jobs WHERE id = $1")
21 .bind(job)
22 .fetch_one(&h.db)
23 .await
24 .expect("status")
25 }
26
27 /// Enqueue one job and claim it (→ running, `started_at` and `heartbeat_at`
28 /// both stamped NOW()).
29 async fn enqueue_and_claim(h: &TestHarness, user: UserId, key: &str) -> Uuid {
30 scan_jobs::enqueue(
31 &h.db,
32 ScanTargetKind::Item,
33 Uuid::new_v4(),
34 key,
35 FileType::Download,
36 user,
37 1000,
38 )
39 .await
40 .expect("enqueue");
41 let job = scan_jobs::claim_next(&h.db)
42 .await
43 .expect("claim")
44 .expect("a queued job to claim");
45 assert_eq!(job.status, "running");
46 assert!(job.heartbeat_at.is_some(), "claim stamps heartbeat_at");
47 job.id
48 }
49
50 #[tokio::test]
51 async fn reaper_spares_slow_but_alive_job() {
52 let h = TestHarness::new().await;
53 let user = seed_user(&h.db, "reaper_alive").await;
54 let job = enqueue_and_claim(&h, user, "k/alive").await;
55
56 // A job running far past the stuck window but whose worker is still beating:
57 // old started_at, fresh heartbeat. This is the exact case that used to be
58 // reaped and double-processed.
59 sqlx::query(
60 "UPDATE scan_jobs SET started_at = NOW() - interval '1 hour', heartbeat_at = NOW() WHERE id = $1",
61 )
62 .bind(job)
63 .execute(&h.db)
64 .await
65 .expect("backdate started, fresh beat");
66
67 let reaped = scan_jobs::reap_stuck(&h.db, 300).await.expect("reap");
68 assert_eq!(reaped, 0, "a job with a fresh heartbeat must not be reaped");
69 assert_eq!(status_of(&h, job).await, "running");
70 }
71
72 #[tokio::test]
73 async fn reaper_reaps_crashed_worker() {
74 let h = TestHarness::new().await;
75 let user = seed_user(&h.db, "reaper_dead").await;
76 let job = enqueue_and_claim(&h, user, "k/dead").await;
77
78 // Worker crashed: no more beats.
79 sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - interval '1 hour' WHERE id = $1")
80 .bind(job)
81 .execute(&h.db)
82 .await
83 .expect("stale beat");
84
85 let reaped = scan_jobs::reap_stuck(&h.db, 300).await.expect("reap");
86 assert_eq!(reaped, 1, "a stale-heartbeat job must be reaped");
87 // attempts after a single claim = 1 < MAX_SCAN_ATTEMPTS, so it returns to
88 // the queue for another attempt rather than being retired to failed.
89 assert_eq!(status_of(&h, job).await, "queued");
90 }
91
92 #[tokio::test]
93 async fn reaper_falls_back_to_started_at_when_no_heartbeat() {
94 // Any row with heartbeat_at NULL (e.g. one in flight across the migration)
95 // stays reapable on the old started_at clock via COALESCE.
96 let h = TestHarness::new().await;
97 let user = seed_user(&h.db, "reaper_null").await;
98 let job = enqueue_and_claim(&h, user, "k/null").await;
99
100 sqlx::query(
101 "UPDATE scan_jobs SET heartbeat_at = NULL, started_at = NOW() - interval '1 hour' WHERE id = $1",
102 )
103 .bind(job)
104 .execute(&h.db)
105 .await
106 .expect("null beat, old start");
107
108 let reaped = scan_jobs::reap_stuck(&h.db, 300).await.expect("reap");
109 assert_eq!(reaped, 1, "null heartbeat falls back to started_at");
110 }
111
112 #[tokio::test]
113 async fn bump_heartbeat_refreshes_and_never_resurrects() {
114 let h = TestHarness::new().await;
115 let user = seed_user(&h.db, "reaper_bump").await;
116 let job = enqueue_and_claim(&h, user, "k/bump").await;
117
118 // A stale beat would be reaped...
119 sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() - interval '1 hour' WHERE id = $1")
120 .bind(job)
121 .execute(&h.db)
122 .await
123 .expect("stale");
124 // ...but a bump refreshes it, so the reaper leaves it alone.
125 scan_jobs::bump_heartbeat(&h.db, job).await.expect("bump");
126 let reaped = scan_jobs::reap_stuck(&h.db, 300).await.expect("reap");
127 assert_eq!(reaped, 0, "bump_heartbeat must refresh liveness");
128 assert_eq!(status_of(&h, job).await, "running");
129
130 // A late bump after the job leaves `running` is a no-op, it must never
131 // resurrect a terminal row's heartbeat.
132 sqlx::query("UPDATE scan_jobs SET status = 'done', heartbeat_at = NULL WHERE id = $1")
133 .bind(job)
134 .execute(&h.db)
135 .await
136 .expect("mark done");
137 scan_jobs::bump_heartbeat(&h.db, job)
138 .await
139 .expect("late bump");
140 let still_null: bool =
141 sqlx::query_scalar("SELECT heartbeat_at IS NULL FROM scan_jobs WHERE id = $1")
142 .bind(job)
143 .fetch_one(&h.db)
144 .await
145 .expect("hb null check");
146 assert!(
147 still_null,
148 "bump must not touch a job that has left running"
149 );
150 }
151