Skip to main content

max / makenotwork

Remediate audit Run 22 fix-first findings; bump to 0.10.8 - auth: unify password length cap on char count across all six sites (login/OAuth/SyncKit switched from byte cap), fixing a permanent self-lockout for multibyte passwords. Shared validation::password_too_long. - scanning: add scan_jobs.heartbeat_at (migration 167); worker beats every 30s and reap_stuck/stuck_count key off COALESCE(heartbeat_at, started_at), so a slow-but-progressing scan is no longer reaped and double-processed. - stripe webhook: route the four charged-but-undelivered WAM escalations through the shutdown-drained bg pool instead of raw tokio::spawn so a ticket survives a mid-deploy restart. - health: gate the live subsystem fan-out behind is_admin (anon gets the cached monitor snapshot), bound the S3 probe at 5s, reconcile docs. Tests: 6 new regression tests (password unit + multibyte login, 4-test scan-job reaper layer suite, health admin-gate); ~300 tests across touched suites green; clean clippy.
Co-Authored-By
Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-04 21:42 UTC
Signed with PGP, not checked
Commit: bfd4533c17f47fb7627243e74b45429de88e0973
Parent: e6502a7
19 files changed, +471 insertions, -71 deletions
M server/Cargo.lock +1 -1
@@ -4318,7 +4318,7 @@
4318 4318
4319 4319 [[package]]
4320 4320 name = "makenotwork"
4321 - version = "0.10.7"
4321 + version = "0.10.8"
4322 4322 dependencies = [
4323 4323 "ammonia",
4324 4324 "anyhow",
M server/Cargo.toml +1 -1
@@ -1,6 +1,6 @@
1 1 [package]
2 2 name = "makenotwork"
3 - version = "0.10.7"
3 + version = "0.10.8"
4 4 edition = "2024"
5 5 license-file = "LICENSE"
6 6
M server/src/db/mod.rs +1 -1
@@ -38,7 +38,7 @@
38 38 pub(crate) mod health;
39 39 pub(crate) mod monitor;
40 40 pub(crate) mod scanning;
41 - pub(crate) mod scan_jobs;
41 + pub mod scan_jobs; // pub so the integration test crate can exercise the reaper/heartbeat layer directly
42 42 pub(crate) mod scan_admin_actions;
43 43 pub(crate) mod content_insertions;
44 44 pub mod invites; // pub so the integration test crate can exercise redemption races directly
M server/src/db/scan_jobs.rs +41 -10
@@ -119,6 +119,11 @@
119 119 pub attempts: i32,
120 120 pub enqueued_at: DateTime<Utc>,
121 121 pub started_at: Option<DateTime<Utc>>,
122 + /// Last liveness beat from the worker running this job. Bumped on a cadence
123 + /// well under STUCK_JOB_SECS; the reaper keys off this (falling back to
124 + /// `started_at`) so a slow-but-progressing scan isn't mistaken for a
125 + /// crashed worker. See `bump_heartbeat` and migration 167.
126 + pub heartbeat_at: Option<DateTime<Utc>>,
122 127 pub completed_at: Option<DateTime<Utc>>,
123 128 pub last_error: Option<String>,
124 129 }
@@ -191,7 +196,8 @@
191 196 UPDATE scan_jobs
192 197 SET status = 'running',
193 198 attempts = attempts + 1,
194 - started_at = NOW()
199 + started_at = NOW(),
200 + heartbeat_at = NOW()
195 201 WHERE id = (SELECT id FROM next)
196 202 RETURNING *
197 203 "#,
@@ -240,14 +246,36 @@
240 246 Ok(())
241 247 }
242 248
243 - /// Reset jobs that have been stuck in `running` longer than `max_age_secs`.
249 + /// Refresh a running job's liveness heartbeat.
250 + ///
251 + /// Called periodically by the worker that owns the job so `reap_stuck` (and the
252 + /// PoM stuck-count) can distinguish a slow-but-progressing scan (fresh beat)
253 + /// from a crashed or hung worker (stale beat). The `status = 'running'` guard
254 + /// makes a late beat a no-op once the job has already finished or been reaped —
255 + /// it never resurrects a terminal row's timestamp.
256 + #[tracing::instrument(skip_all, fields(%job_id))]
257 + pub async fn bump_heartbeat(db: &PgPool, job_id: Uuid) -> Result<(), sqlx::Error> {
258 + sqlx::query("UPDATE scan_jobs SET heartbeat_at = NOW() WHERE id = $1 AND status = 'running'")
259 + .bind(job_id)
260 + .execute(db)
261 + .await?;
262 + Ok(())
263 + }
264 +
265 + /// Reset jobs whose worker has gone silent for longer than `max_age_secs`.
244 266 ///
245 267 /// Run on worker startup and on a timer to recover from a previous-process crash
246 268 /// mid-scan: the row would otherwise stay `running` forever and never be
247 - /// re-claimed. A job whose `attempts` has already reached `MAX_SCAN_ATTEMPTS` is
248 - /// retired to `failed` (each claim increments `attempts`, so a scan that
249 - /// reliably wedges the process trips the budget); otherwise it returns to
250 - /// `queued` for another attempt.
269 + /// re-claimed. Liveness is measured from the job's last heartbeat
270 + /// (`COALESCE(heartbeat_at, started_at)`), NOT from `started_at` alone — a
271 + /// legitimately slow scan (large object over a slow S3 link) can run past
272 + /// `max_age_secs` while still making progress, and reaping *that* let a second
273 + /// worker double-process the same object and inflate `attempts` toward
274 + /// MAX_SCAN_ATTEMPTS until a valid file was force-retired (audit Run 22). The
275 + /// running worker bumps `heartbeat_at` on a cadence well under `max_age_secs`,
276 + /// so only a crashed/hung process (no fresh beat) crosses the threshold. A job
277 + /// whose `attempts` has already reached `MAX_SCAN_ATTEMPTS` is retired to
278 + /// `failed`; otherwise it returns to `queued` for another attempt.
251 279 #[tracing::instrument(skip_all)]
252 280 pub async fn reap_stuck(db: &PgPool, max_age_secs: i64) -> Result<u64, sqlx::Error> {
253 281 let affected = sqlx::query(
@@ -255,12 +283,13 @@
255 283 UPDATE scan_jobs
256 284 SET status = CASE WHEN attempts >= $2 THEN 'failed' ELSE 'queued' END,
257 285 started_at = NULL,
286 + heartbeat_at = NULL,
258 287 completed_at = CASE WHEN attempts >= $2 THEN NOW() ELSE completed_at END,
259 288 last_error = CASE WHEN attempts >= $2
260 289 THEN 'exceeded max scan attempts (reaped from running)'
261 290 ELSE last_error END
262 291 WHERE status = 'running'
263 - AND started_at < NOW() - ($1 || ' seconds')::interval
292 + AND COALESCE(heartbeat_at, started_at) < NOW() - ($1 || ' seconds')::interval
264 293 "#,
265 294 )
266 295 .bind(max_age_secs.to_string())
@@ -318,11 +347,13 @@
318 347 .await
319 348 }
320 349
321 - /// Count of running jobs that have been in flight longer than `max_age_secs`.
322 - /// Used by PoM to alert on stuck workers.
350 + /// Count of running jobs whose worker has gone silent longer than
351 + /// `max_age_secs`. Used by PoM to alert on stuck workers. Keyed off the
352 + /// heartbeat (falling back to `started_at`) so a slow-but-alive scan isn't
353 + /// counted as stuck — matching `reap_stuck`'s liveness definition.
323 354 pub async fn stuck_count(db: &PgPool, max_age_secs: i64) -> Result<i64, sqlx::Error> {
324 355 sqlx::query_scalar::<_, i64>(
325 - "SELECT COUNT(*) FROM scan_jobs WHERE status = 'running' AND started_at < NOW() - ($1 || ' seconds')::interval",
356 + "SELECT COUNT(*) FROM scan_jobs WHERE status = 'running' AND COALESCE(heartbeat_at, started_at) < NOW() - ($1 || ' seconds')::interval",
326 357 )
327 358 .bind(max_age_secs.to_string())
328 359 .fetch_one(db)
M server/src/routes/auth.rs +4 -2
@@ -177,8 +177,10 @@
177 177 ));
178 178 }
179 179
180 - // Cap password length to match signup validation (prevents Argon2 DoS with huge inputs)
181 - if form.password.len() > 128 {
180 + // Cap password length to match signup validation (prevents Argon2 DoS with
181 + // huge inputs). MUST use the same char-count metric as signup, or a valid
182 + // multibyte password (<=128 chars, >128 bytes) is silently rejected here.
183 + if crate::validation::password_too_long(&form.password) {
182 184 return return_error("Invalid username/email or password");
183 185 }
184 186
M server/src/routes/oauth.rs +4 -2
@@ -498,8 +498,10 @@
498 498 ));
499 499 }
500 500
501 - // Cap password length to prevent DoS via Argon2 on very long inputs
502 - if password.len() > 128 {
501 + // Cap password length to prevent DoS via Argon2 on very long inputs.
502 + // Same char-count metric as signup (bytes would lock out multibyte
503 + // passwords that were accepted at signup).
504 + if crate::validation::password_too_long(password) {
503 505 return Ok(render_authorize_error(
504 506 Some(csrf_token),
505 507 session_user,
M server/src/scanning/worker.rs +48 -4
@@ -24,14 +24,58 @@
24 24 /// Worker poll interval when the queue is empty.
25 25 const IDLE_POLL_INTERVAL: Duration = Duration::from_millis(500);
26 26
27 - /// How long a `running` job can sit before the reaper resets it to `queued`.
28 - /// A scan that legitimately takes longer than this is an outlier and warrants
29 - /// admin attention anyway.
27 + /// How long a `running` job can go without a heartbeat before the reaper resets
28 + /// it. This is measured from the worker's last liveness beat (see
29 + /// [`HEARTBEAT_INTERVAL`]), not from claim time, so a genuinely slow-but-
30 + /// progressing scan is never reaped — only a crashed or hung worker crosses it.
30 31 const STUCK_JOB_SECS: i64 = 300;
31 32
33 + /// Cadence at which the worker running a job refreshes its `heartbeat_at`. Must
34 + /// be comfortably smaller than [`STUCK_JOB_SECS`] (here ~10x) so a live job that
35 + /// is merely slow keeps beating well inside the reaper's window.
36 + const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(30);
37 +
32 38 /// Cadence at which any worker tries to reap stuck jobs.
33 39 const REAPER_INTERVAL: Duration = Duration::from_secs(60);
34 40
41 + /// Aborts a spawned task when dropped, so a job's heartbeat companion never
42 + /// outlives the job — including when `process_job` returns early with an error.
43 + struct AbortOnDrop(tokio::task::JoinHandle<()>);
44 +
45 + impl Drop for AbortOnDrop {
46 + fn drop(&mut self) {
47 + self.0.abort();
48 + }
49 + }
50 +
51 + /// Run `process_job` while a companion task keeps the job's `heartbeat_at`
52 + /// fresh. The reaper distinguishes a slow-but-alive scan (recent beat) from a
53 + /// crashed worker (stale beat) purely by this heartbeat, so a large scan that
54 + /// runs past `STUCK_JOB_SECS` is no longer reclaimed and double-processed. The
55 + /// companion is aborted the moment the job finishes (via `AbortOnDrop`),
56 + /// error or not, so it cannot bump a job that has already left `running`.
57 + async fn process_job_with_heartbeat(
58 + ctx: &WorkerContext,
59 + job: ScanJob,
60 + ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
61 + let job_id = job.id;
62 + let hb_db = ctx.db.clone();
63 + let heartbeat = tokio::spawn(async move {
64 + let mut ticker = tokio::time::interval(HEARTBEAT_INTERVAL);
65 + // The claim already stamped heartbeat_at, so skip the immediate first
66 + // tick and beat one interval from now.
67 + ticker.tick().await;
68 + loop {
69 + ticker.tick().await;
70 + if let Err(e) = db::scan_jobs::bump_heartbeat(&hb_db, job_id).await {
71 + tracing::warn!(%job_id, error = %e, "scan heartbeat bump failed");
72 + }
73 + }
74 + });
75 + let _guard = AbortOnDrop(heartbeat);
76 + process_job(ctx, job).await
77 + }
78 +
35 79 /// Shared dependencies the worker pool needs.
36 80 pub struct WorkerContext {
37 81 pub db: PgPool,
@@ -116,7 +160,7 @@
116 160 match db::scan_jobs::claim_next(&ctx.db).await {
117 161 Ok(Some(job)) => {
118 162 let job_id = job.id;
119 - if let Err(e) = process_job(&ctx, job).await {
163 + if let Err(e) = process_job_with_heartbeat(&ctx, job).await {
120 164 tracing::error!(worker_id, %job_id, error = %e, "scan job failed");
121 165 if let Err(e2) = db::scan_jobs::mark_failed(&ctx.db, job_id, &e.to_string()).await {
122 166 tracing::error!(worker_id, %job_id, error = %e2, "failed to mark job failed");
M server/src/validation/mod.rs +39
M server/tests/workflows/admin.rs +28
M server/tests/workflows/auth.rs +31
M server/tests/workflows/mod.rs +1
M server/src/routes/synckit/auth.rs +3 -2
M server/src/routes/api/users/profile.rs +2 -2
M server/src/routes/pages/email_actions/password.rs +2 -2
M server/src/routes/pages/public/join_wizard.rs +2 -2
M server/src/routes/stripe/webhook/checkout.rs +17 -7
M server/src/routes/pages/public/health/mod.rs +66 -35
A server/migrations/167_scan_jobs_heartbeat.sql +22
A server/tests/workflows/db_scan_jobs_layer.rs +158