Skip to main content

max / makenotwork

server: import-job heartbeat + stuck-job reaper, typed status (A5) Imports run on the bounded background pool with no worker loop, so a crash mid-import stranded the job in `processing` forever — the creator's dashboard showed it running indefinitely with no way to retry, and it was untested. - Migration 168 adds import_jobs.heartbeat_at + a partial index for the reaper. - update_import_status is now typed on ImportJobStatus (not a raw &str) and stamps heartbeat_at when a job enters `processing`; bump_import_heartbeat refreshes it after every chunk. - reap_stuck_import_jobs fails jobs whose heartbeat has gone stale past the threshold; wired into the scheduler's hourly tick (STUCK_IMPORT_SECS = 30min, comfortably above the per-chunk beat cadence). - import_tiers now matches the typed SQLSTATE via helpers::is_unique_violation instead of substring-scanning the error Display for "23505" (A6). - New db_imports_layer integration tests cover CRUD, user-scoping, the Processing->heartbeat stamp, reaper-fails-stale, reaper-spares-fresh/terminal, and bump-only-touches-processing. db::imports is now pub for the test crate (mirrors db::scan_jobs). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Author: Max Johnson <me@maxj.phd> · 2026-07-07 21:27 UTC
Commit: 7af46fb78a9158a9ab78dd91fd03cd69fb345e49
Parent: 5bec81b
7 files changed, +269 insertions, -10 deletions
A server/migrations/168_import_jobs_heartbeat.sql +15
@@ -0,0 +1,15 @@
1 + -- Import-job liveness heartbeat, mirroring scan_jobs (migration 167).
2 + --
3 + -- An import runs on the bounded background pool, not a dedicated worker loop,
4 + -- and had no liveness signal: a server crash mid-import stranded the job in
5 + -- `processing` forever (the dashboard poller would show it running indefinitely
6 + -- and the creator could never retry). `heartbeat_at` is stamped when the job
7 + -- enters `processing` and bumped after every chunk; a scheduler reaper fails
8 + -- jobs whose heartbeat has gone stale.
9 + ALTER TABLE import_jobs ADD COLUMN heartbeat_at TIMESTAMPTZ;
10 +
11 + -- Partial index for the reaper's `status = 'processing' AND heartbeat_at < …`
12 + -- sweep — only in-flight jobs are ever scanned.
13 + CREATE INDEX idx_import_jobs_processing_heartbeat
14 + ON import_jobs (heartbeat_at)
15 + WHERE status = 'processing';
M server/src/db/imports.rs +56 -3
@@ -2,6 +2,7 @@
2 2
3 3 use sqlx::PgPool;
4 4
5 + use super::enums::ImportJobStatus;
5 6 use super::id_types::{ImportJobId, ProjectId, UserId};
6 7 use super::models::DbImportJob;
7 8 use crate::import::ImportSource;
@@ -60,21 +61,73 @@ pub async fn update_import_progress(
60 61 }
61 62
62 63 /// Update the status of an import job.
64 + ///
65 + /// Typed on [`ImportJobStatus`] rather than a raw `&str` so a caller can't stamp
66 + /// a status the reader ([`DbImportJob::status`]) can't decode. Entering
67 + /// `Processing` also stamps `heartbeat_at = NOW()` so the reaper has a liveness
68 + /// baseline from the moment the job starts running (the first chunk may take a
69 + /// while); other transitions leave the heartbeat untouched.
63 70 #[tracing::instrument(skip_all)]
64 71 pub async fn update_import_status(
65 72 pool: &PgPool,
66 73 job_id: ImportJobId,
67 - status: &str,
74 + status: ImportJobStatus,
68 75 ) -> Result<()> {
69 - sqlx::query("UPDATE import_jobs SET status = $2 WHERE id = $1")
76 + sqlx::query(
77 + "UPDATE import_jobs \
78 + SET status = $2, \
79 + heartbeat_at = CASE WHEN $2 = 'processing' THEN NOW() ELSE heartbeat_at END \
80 + WHERE id = $1",
81 + )
82 + .bind(job_id)
83 + .bind(status)
84 + .execute(pool)
85 + .await?;
86 +
87 + Ok(())
88 + }
89 +
90 + /// Refresh a running import's liveness heartbeat. Called after every processed
91 + /// chunk so [`reap_stuck_import_jobs`] can tell a slow-but-progressing import
92 + /// (fresh beat) from one whose owning process died (stale beat).
93 + #[tracing::instrument(skip_all)]
94 + pub async fn bump_import_heartbeat(pool: &PgPool, job_id: ImportJobId) -> Result<()> {
95 + sqlx::query("UPDATE import_jobs SET heartbeat_at = NOW() WHERE id = $1 AND status = 'processing'")
70 96 .bind(job_id)
71 - .bind(status)
72 97 .execute(pool)
73 98 .await?;
74 99
75 100 Ok(())
76 101 }
77 102
103 + /// Fail every import job stuck in `processing` whose heartbeat has gone stale
104 + /// past `max_age_secs` — i.e. the process running it died without reaching
105 + /// `complete_import_job` / `fail_import_job`. Without this a crash mid-import
106 + /// leaves the job `processing` forever. Returns the number of jobs reaped.
107 + ///
108 + /// `COALESCE(heartbeat_at, created_at)` covers a job that crashed before its
109 + /// first heartbeat (older rows and the window between insert and the first
110 + /// `processing` stamp). Mirrors `scan_jobs::reap_stuck`'s liveness definition.
111 + #[tracing::instrument(skip_all)]
112 + pub async fn reap_stuck_import_jobs(pool: &PgPool, max_age_secs: i64) -> Result<u64> {
113 + let affected = sqlx::query(
114 + "UPDATE import_jobs \
115 + SET status = 'failed', \
116 + completed_at = NOW(), \
117 + error_log = COALESCE(error_log, '') || \
118 + CASE WHEN error_log IS NULL OR error_log = '' THEN '' ELSE E'\\n' END || \
119 + 'import reaped: no heartbeat for over the stuck-job threshold (process likely crashed)' \
120 + WHERE status = 'processing' \
121 + AND COALESCE(heartbeat_at, created_at) < NOW() - ($1 || ' seconds')::interval",
122 + )
123 + .bind(max_age_secs.to_string())
124 + .execute(pool)
125 + .await?
126 + .rows_affected();
127 +
128 + Ok(affected)
129 + }
130 +
78 131 /// Mark an import job as completed with optional error log.
79 132 #[tracing::instrument(skip_all)]
80 133 pub async fn complete_import_job(
M server/src/db/mod.rs +1 -1
@@ -60,7 +60,7 @@ pub mod custom_domains;
60 60 pub mod patches;
61 61 pub mod bundles;
62 62 pub(crate) mod email_signups;
63 - pub(crate) mod imports;
63 + pub mod imports; // pub so the integration test crate can exercise the reaper/heartbeat layer directly
64 64 pub(crate) mod media_files;
65 65 pub mod tips;
66 66 pub(crate) mod project_members;
M server/src/import/pipeline.rs +13 -6
@@ -9,7 +9,7 @@
9 9 use sqlx::PgPool;
10 10
11 11 use super::{ImportPayload, ImportTier};
12 - use crate::db::{self, ImportJobId, ItemType, PriceCents, ProjectId, UserId};
12 + use crate::db::{self, ImportJobId, ImportJobStatus, ItemType, PriceCents, ProjectId, UserId};
13 13 use crate::error::Result;
14 14
15 15 /// Chunk size for progress updates.
@@ -36,7 +36,7 @@ pub async fn run_import(
36 36 _user_id: UserId,
37 37 payload: ImportPayload,
38 38 ) -> Result<()> {
39 - db::imports::update_import_status(pool, job_id, "processing").await?;
39 + db::imports::update_import_status(pool, job_id, ImportJobStatus::Processing).await?;
40 40
41 41 let mut processed: i32 = 0;
42 42 let mut created: i32 = 0;
@@ -167,9 +167,12 @@ async fn import_tiers(
167 167 {
168 168 Ok(_) => count += 1,
169 169 Err(e) => {
170 - // Check for duplicate (name already exists for project)
171 - if e.to_string().contains("23505") {
172 - // Unique violation — tier already exists, skip
170 + // A unique violation (23505) means the tier name already exists
171 + // for this project — an idempotent skip, not an error. Match on
172 + // the typed SQLSTATE rather than substring-scanning the Display
173 + // string (which could false-match a code appearing in a message).
174 + if crate::helpers::is_unique_violation(&e) {
175 + // Tier already exists — skip.
173 176 } else {
174 177 errors.push(format!("Tier '{}': {}", tier.name, e));
175 178 }
@@ -292,11 +295,15 @@ fn strip_html_tags(html: &str) -> String {
292 295 }
293 296 }
294 297
295 - /// Update progress on the import job.
298 + /// Update progress on the import job, and refresh its liveness heartbeat so the
299 + /// reaper doesn't mistake a long-but-progressing import for a crashed one.
296 300 async fn update_progress(pool: &PgPool, job_id: ImportJobId, processed: i32, created: i32, skipped: i32) {
297 301 if let Err(e) = db::imports::update_import_progress(pool, job_id, processed, created, skipped).await {
298 302 tracing::warn!(error = %e, "failed to update import progress");
299 303 }
304 + if let Err(e) = db::imports::bump_import_heartbeat(pool, job_id).await {
305 + tracing::warn!(error = %e, "failed to bump import heartbeat");
306 + }
300 307 }
301 308
302 309 #[cfg(test)]
M server/src/scheduler/mod.rs +12
@@ -62,6 +62,11 @@ const TICK_DURATION_ALERT_SECS: u64 = 50;
62 62 /// healthy-tick ceiling, so it only trips on a true wedge.
63 63 const TICK_WATCHDOG_SECS: u64 = 300;
64 64
65 + /// An import job whose liveness heartbeat is older than this is treated as
66 + /// crashed and failed by the hourly reaper. Heartbeats bump after every 50-item
67 + /// chunk, so 30 minutes of silence means the owning process is gone, not slow.
68 + const STUCK_IMPORT_SECS: i64 = 1800;
69 +
65 70 /// Determine which scheduled job groups should run for a given tick.
66 71 ///
67 72 /// Returns `(sandbox_cleanup, hourly_jobs, daily_jobs, weekly_jobs)`.
@@ -272,6 +277,13 @@ pub fn spawn_scheduler(
272 277 if run_hourly {
273 278 synckit_warnings::check_and_send_warnings(&state).await;
274 279 cleanup::purge_old_scan_jobs(&state).await;
280 + // Fail imports stranded in `processing` by a crashed process.
281 + // Heartbeats bump per chunk, so a 30-min-stale beat is dead.
282 + match db::imports::reap_stuck_import_jobs(&state.db, STUCK_IMPORT_SECS).await {
283 + Ok(n) if n > 0 => tracing::warn!(reaped = n, "failed stuck import jobs (stale heartbeat)"),
284 + Ok(_) => {}
285 + Err(e) => tracing::error!(error = ?e, "failed to reap stuck import jobs"),
286 + }
275 287 }
276 288
277 289 // Weekly storage drift correction + integrity checks
A server/tests/workflows/db_imports_layer.rs +171
@@ -0,0 +1,171 @@
1 + //! DB-layer contract tests for `db/imports.rs`: the import-job CRUD surface, the
2 + //! typed status transitions, the liveness heartbeat, and the stuck-job reaper.
3 + //!
4 + //! Audit A5: imports run on the bounded background pool with no worker loop, so
5 + //! before migration 168 a crash mid-import stranded the job in `processing`
6 + //! forever. These pin that `update_import_status(Processing)` seeds a heartbeat,
7 + //! `bump_import_heartbeat` refreshes it, and `reap_stuck_import_jobs` fails a
8 + //! stale-heartbeat job while sparing a freshly-beating or already-terminal one.
9 +
10 + use crate::harness::TestHarness;
11 + use makenotwork::db::{self, ImportJobStatus, ProjectId};
12 + use makenotwork::import::ImportSource;
13 +
14 + /// Create a creator + project and return (user_id, project_id) typed for direct
15 + /// `db::imports` calls.
16 + async fn seed_creator(h: &mut TestHarness) -> (makenotwork::db::UserId, ProjectId) {
17 + let setup = h.create_creator_with_item("importlayer", "digital", 0).await;
18 + let project_id: ProjectId = setup.project_id.parse().expect("project_id parses");
19 + (setup.user_id, project_id)
20 + }
21 +
22 + #[tokio::test]
23 + async fn import_job_crud_roundtrip() {
24 + let mut h = TestHarness::new().await;
25 + let (user_id, project_id) = seed_creator(&mut h).await;
26 +
27 + let job = db::imports::create_import_job(&h.db, user_id, project_id, ImportSource::GenericCsv, 10)
28 + .await
29 + .expect("create job");
30 + assert_eq!(job.status, ImportJobStatus::Pending);
31 + assert_eq!(job.total_rows, 10);
32 +
33 + db::imports::update_import_progress(&h.db, job.id, 5, 4, 1).await.expect("progress");
34 + let fetched = db::imports::get_import_job(&h.db, job.id, user_id)
35 + .await
36 + .expect("get")
37 + .expect("job exists");
38 + assert_eq!(fetched.processed_rows, 5);
39 + assert_eq!(fetched.created_rows, 4);
40 + assert_eq!(fetched.skipped_rows, 1);
41 +
42 + db::imports::complete_import_job(&h.db, job.id, Some("2 warnings".into()))
43 + .await
44 + .expect("complete");
45 + let done = db::imports::get_import_job(&h.db, job.id, user_id)
46 + .await
47 + .expect("get")
48 + .expect("job exists");
49 + assert_eq!(done.status, ImportJobStatus::Completed);
50 + assert_eq!(done.error_log.as_deref(), Some("2 warnings"));
51 + assert!(done.completed_at.is_some());
52 +
53 + // Scoping: another user can't read the job.
54 + let other = h.signup("importlayer2", "il2@example.com", "Password1!").await;
55 + assert!(
56 + db::imports::get_import_job(&h.db, job.id, other).await.unwrap().is_none(),
57 + "get_import_job must be user-scoped"
58 + );
59 + }
60 +
61 + #[tokio::test]
62 + async fn processing_status_seeds_heartbeat() {
63 + let mut h = TestHarness::new().await;
64 + let (user_id, project_id) = seed_creator(&mut h).await;
65 + let job = db::imports::create_import_job(&h.db, user_id, project_id, ImportSource::GenericCsv, 1)
66 + .await
67 + .unwrap();
68 +
69 + // Pending job has no heartbeat.
70 + let hb0: Option<chrono::DateTime<chrono::Utc>> =
71 + sqlx::query_scalar("SELECT heartbeat_at FROM import_jobs WHERE id = $1")
72 + .bind(job.id)
73 + .fetch_one(&h.db)
74 + .await
75 + .unwrap();
76 + assert!(hb0.is_none(), "a pending job must not have a heartbeat");
77 +
78 + db::imports::update_import_status(&h.db, job.id, ImportJobStatus::Processing).await.unwrap();
79 + let hb1: Option<chrono::DateTime<chrono::Utc>> =
80 + sqlx::query_scalar("SELECT heartbeat_at FROM import_jobs WHERE id = $1")
81 + .bind(job.id)
82 + .fetch_one(&h.db)
83 + .await
84 + .unwrap();
85 + assert!(hb1.is_some(), "entering Processing must stamp a heartbeat");
86 + }
87 +
88 + #[tokio::test]
89 + async fn reaper_fails_stale_processing_job() {
90 + let mut h = TestHarness::new().await;
91 + let (user_id, project_id) = seed_creator(&mut h).await;
92 + let job = db::imports::create_import_job(&h.db, user_id, project_id, ImportSource::GenericCsv, 100)
93 + .await
94 + .unwrap();
95 + db::imports::update_import_status(&h.db, job.id, ImportJobStatus::Processing).await.unwrap();
96 +
97 + // Owning process crashed: heartbeat goes stale.
98 + sqlx::query("UPDATE import_jobs SET heartbeat_at = NOW() - interval '1 hour' WHERE id = $1")
99 + .bind(job.id)
100 + .execute(&h.db)
101 + .await
102 + .unwrap();
103 +
104 + let reaped = db::imports::reap_stuck_import_jobs(&h.db, 1800).await.unwrap();
105 + assert_eq!(reaped, 1, "a stale-heartbeat processing job must be reaped");
106 +
107 + let after = db::imports::get_import_job(&h.db, job.id, user_id).await.unwrap().unwrap();
108 + assert_eq!(after.status, ImportJobStatus::Failed);
109 + assert!(after.completed_at.is_some());
110 + assert!(
111 + after.error_log.as_deref().unwrap_or("").contains("reaped"),
112 + "reaped job error_log should explain why: {:?}",
113 + after.error_log
114 + );
115 + }
116 +
117 + #[tokio::test]
118 + async fn reaper_spares_fresh_and_terminal_jobs() {
119 + let mut h = TestHarness::new().await;
120 + let (user_id, project_id) = seed_creator(&mut h).await;
121 +
122 + // Fresh, actively-beating processing job.
123 + let fresh = db::imports::create_import_job(&h.db, user_id, project_id, ImportSource::GenericCsv, 100)
124 + .await
125 + .unwrap();
126 + db::imports::update_import_status(&h.db, fresh.id, ImportJobStatus::Processing).await.unwrap();
127 + db::imports::bump_import_heartbeat(&h.db, fresh.id).await.unwrap();
128 +
129 + // Already-completed job — terminal, must never be reaped.
130 + let done = db::imports::create_import_job(&h.db, user_id, project_id, ImportSource::GenericCsv, 1)
131 + .await
132 + .unwrap();
133 + db::imports::complete_import_job(&h.db, done.id, None).await.unwrap();
134 +
135 + let reaped = db::imports::reap_stuck_import_jobs(&h.db, 1800).await.unwrap();
136 + assert_eq!(reaped, 0, "a fresh-heartbeat job and a completed job must be spared");
137 +
138 + assert_eq!(
139 + db::imports::get_import_job(&h.db, fresh.id, user_id).await.unwrap().unwrap().status,
140 + ImportJobStatus::Processing
141 + );
142 + assert_eq!(
143 + db::imports::get_import_job(&h.db, done.id, user_id).await.unwrap().unwrap().status,
144 + ImportJobStatus::Completed
145 + );
146 + }
147 +
148 + #[tokio::test]
149 + async fn bump_heartbeat_only_touches_processing_jobs() {
150 + let mut h = TestHarness::new().await;
151 + let (user_id, project_id) = seed_creator(&mut h).await;
152 +
153 + // A completed (terminal) job must not be revived to a beating state.
154 + let done = db::imports::create_import_job(&h.db, user_id, project_id, ImportSource::GenericCsv, 1)
155 + .await
156 + .unwrap();
157 + db::imports::complete_import_job(&h.db, done.id, None).await.unwrap();
158 + db::imports::bump_import_heartbeat(&h.db, done.id).await.unwrap();
159 +
160 + let hb: Option<chrono::DateTime<chrono::Utc>> =
161 + sqlx::query_scalar("SELECT heartbeat_at FROM import_jobs WHERE id = $1")
162 + .bind(done.id)
163 + .fetch_one(&h.db)
164 + .await
165 + .unwrap();
166 + assert!(hb.is_none(), "bump must not stamp a heartbeat on a terminal job");
167 + assert_eq!(
168 + db::imports::get_import_job(&h.db, done.id, user_id).await.unwrap().unwrap().status,
169 + ImportJobStatus::Completed
170 + );
171 + }
M server/tests/workflows/mod.rs +1
@@ -97,6 +97,7 @@ mod lifecycle;
97 97 mod cart;
98 98 mod bundles;
99 99 mod idempotency;
100 + mod db_imports_layer;
100 101 mod db_items_layer;
101 102 mod db_scan_jobs_layer;
102 103 mod db_payments_layer;