cortex/
jobs.rs

1// Copyright 2015-2025 Deyan Ginev. See the LICENSE
2// file at the top-level directory of this distribution.
3//
4// Licensed under the MIT license <LICENSE-MIT or http://opensource.org/licenses/MIT>.
5// This file may not be copied, modified, or distributed
6// except according to those terms.
7
8//! Background jobs: one persisted row per long-running administrative operation, run on an
9//! in-process thread with progress persisted to the database. The shared mechanism behind corpus
10//! import/extend, service activation, runs, and dataset export. See `docs/archive/JOB_MODEL.md`.
11
12use std::thread;
13
14use chrono::NaiveDateTime;
15use diesel::dsl::now;
16use diesel::pg::PgConnection;
17use diesel::prelude::*;
18use serde_json::Value;
19use uuid::Uuid;
20
21use crate::backend::DbPool;
22use crate::config::config;
23use crate::schema::jobs;
24
25/// A persisted background job.
26#[derive(Queryable, Identifiable, Clone, Debug)]
27#[diesel(table_name = jobs)]
28pub struct Job {
29  /// Internal serial id.
30  pub id: i64,
31  /// External handle.
32  pub uuid: Uuid,
33  /// Operation kind (e.g. `corpus_import`).
34  pub kind: String,
35  /// `queued` | `running` | `succeeded` | `failed` | `interrupted`.
36  pub status: String,
37  /// Units of work completed.
38  pub progress_current: i32,
39  /// Total units of work, when known.
40  pub progress_total: Option<i32>,
41  /// Current step, or the error message on failure.
42  pub message: String,
43  /// Who started the job (Arm 9 identity).
44  pub actor: String,
45  /// Job inputs.
46  pub params: Value,
47  /// Terminal result payload.
48  pub result: Option<Value>,
49  /// When the job was created.
50  pub created_at: NaiveDateTime,
51  /// When the job was last updated.
52  pub updated_at: NaiveDateTime,
53}
54
55#[derive(Insertable)]
56#[diesel(table_name = jobs)]
57struct NewJob {
58  kind: String,
59  actor: String,
60  params: Value,
61}
62
63/// A handle passed to a job body; each call persists progress on the job row.
64pub struct JobProgress {
65  pool: DbPool,
66  job_id: i64,
67}
68impl JobProgress {
69  /// Records progress (`current`/`total` and a human-readable `message`) on the job row.
70  pub fn step(&self, current: i32, total: Option<i32>, message: &str) {
71    if let Ok(mut connection) = self.pool.get() {
72      let _ = diesel::update(jobs::table.filter(jobs::id.eq(self.job_id)))
73        .set((
74          jobs::progress_current.eq(current),
75          jobs::progress_total.eq(total),
76          jobs::message.eq(message),
77          jobs::updated_at.eq(now),
78        ))
79        .execute(&mut connection);
80    }
81  }
82}
83
84/// Spawns a background job: inserts a `queued` row, returns its uuid, and runs `body` on a thread.
85/// The body reports progress via [`JobProgress`] and returns a terminal result or an error message.
86pub fn spawn_job<F>(
87  pool: DbPool,
88  kind: &str,
89  actor: &str,
90  params: Value,
91  body: F,
92) -> Result<Uuid, String>
93where
94  F: FnOnce(&JobProgress) -> Result<Value, String> + Send + 'static,
95{
96  let mut connection = pool.get().map_err(|e| e.to_string())?;
97  let (job_id, job_uuid): (i64, Uuid) = diesel::insert_into(jobs::table)
98    .values(NewJob {
99      kind: kind.to_string(),
100      actor: actor.to_string(),
101      params,
102    })
103    .returning((jobs::id, jobs::uuid))
104    .get_result(&mut connection)
105    .map_err(|e| e.to_string())?;
106  drop(connection);
107
108  // Structured operational journal for the whole job lifecycle (the jobs table is the durable
109  // record; this is the leveled `tracing` stream an operator tails alongside the dispatcher, so
110  // background activity — imports, reruns, reindex — is visible and correlatable in one place).
111  // Captured (owned) for the worker thread, which can't borrow these `&str`s.
112  let log_kind = kind.to_string();
113  let log_actor = actor.to_string();
114  tracing::info!(kind = %log_kind, actor = %log_actor, job = %job_uuid, "background job spawned");
115
116  let worker_pool = pool.clone();
117  thread::spawn(move || {
118    let started = std::time::Instant::now();
119    set_running(&worker_pool, job_id);
120    let progress = JobProgress {
121      pool: worker_pool.clone(),
122      job_id,
123    };
124    // Catch a panicking body so a job is never stranded `running` forever (e.g. an importer panic,
125    // or `from_address`/`connection_at` panicking when the DB is briefly unreachable). A panic
126    // becomes a terminal `failed` with a message — the job reaches a real health state.
127    let (status, message) =
128      match std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| body(&progress))) {
129        Ok(Ok(result)) => {
130          finish(&worker_pool, job_id, "succeeded", "", Some(result));
131          ("succeeded", String::new())
132        },
133        Ok(Err(message)) => {
134          finish(&worker_pool, job_id, "failed", &message, None);
135          ("failed", message)
136        },
137        Err(panic) => {
138          let message = format!("job panicked: {}", panic_message(&*panic));
139          finish(&worker_pool, job_id, "failed", &message, None);
140          ("failed", message)
141        },
142      };
143    let elapsed_ms = started.elapsed().as_millis();
144    if status == "succeeded" {
145      tracing::info!(kind = %log_kind, actor = %log_actor, job = %job_uuid, elapsed_ms, "background job succeeded");
146    } else {
147      tracing::warn!(kind = %log_kind, actor = %log_actor, job = %job_uuid, elapsed_ms, error = %message, "background job failed");
148    }
149  });
150  Ok(job_uuid)
151}
152
153/// The job `kind` for a `report_summary` rollup refresh.
154pub const REFRESH_REPORTS_KIND: &str = "refresh_reports";
155
156/// Spawns a background job for the manual "Force refresh" action. The global `report_summary`
157/// matview was retired in favour of a per-(corpus, service, severity) cache (`report_grain_cache`),
158/// so this no longer does a multi-minute rebuild: it **invalidates** the whole cache (a cheap
159/// `DELETE`, never a scan), and each report slice then repopulates lazily, per scope, on its next
160/// view. Run **off** the request path so the caller returns immediately.
161///
162/// **Debounced:** one global invalidation suffices for every report page, so if a job is already
163/// queued or running its uuid is returned instead of spawning another. Poll `GET /api/jobs/<uuid>`
164/// for status/health.
165pub fn spawn_report_refresh(pool: DbPool, actor: &str) -> Result<Uuid, String> {
166  // Debounce against an already-active refresh before inserting a new job row.
167  {
168    let mut connection = pool.get().map_err(|e| e.to_string())?;
169    if let Some(existing) = list_recent(&mut connection, true, 200)
170      .into_iter()
171      .find(|job| job.kind == REFRESH_REPORTS_KIND)
172    {
173      return Ok(existing.uuid);
174    }
175  }
176  spawn_job(pool, REFRESH_REPORTS_KIND, actor, Value::Null, |progress| {
177    let mut connection = progress.pool.get().map_err(|e| e.to_string())?;
178    crate::backend::invalidate_all(&mut connection).map_err(|e| e.to_string())?;
179    Ok(serde_json::json!({ "status": "cache-invalidated" }))
180  })
181}
182
183/// The job `kind` for a single `(corpus, service, severity)` report-cache populate.
184pub const POPULATE_REPORT_KIND: &str = "populate_report";
185
186/// Whether a populate job for exactly this `(corpus, service, severity)` slice is already queued or
187/// running — the request path's pre-check, so a burst of viewers of a still-computing report
188/// doesn't each re-attempt the (bounded) inline aggregation while the background job is already on
189/// it.
190pub fn report_populate_active(
191  connection: &mut PgConnection,
192  corpus_id: i32,
193  service_id: i32,
194  severity: &str,
195) -> bool {
196  list_recent(connection, true, 200)
197    .into_iter()
198    .any(|job| populate_job_matches(&job, corpus_id, service_id, severity))
199}
200
201fn populate_job_matches(job: &Job, corpus_id: i32, service_id: i32, severity: &str) -> bool {
202  job.kind == POPULATE_REPORT_KIND
203    && job.params.get("corpus_id").and_then(Value::as_i64) == Some(corpus_id as i64)
204    && job.params.get("service_id").and_then(Value::as_i64) == Some(service_id as i64)
205    && job.params.get("severity").and_then(Value::as_str) == Some(severity)
206}
207
208/// Spawns a background job that (re)computes one `(corpus, service, severity)` report slice
209/// (`rollup::populate_scope`) **off** the request path — the heavy aggregation behind a cold report
210/// view (minutes for the full-arXiv `info` slice) runs here instead of pinning a frontend
211/// connection and blocking the viewer, who is shown a "report computing" page that refreshes when
212/// the slice is ready. **Debounced** per scope: an already-active populate for the same slice is
213/// reused (its uuid returned) rather than spawning a duplicate. Poll `GET /api/jobs/<uuid>` for
214/// status/health.
215pub fn spawn_report_populate(
216  pool: DbPool,
217  corpus_id: i32,
218  service_id: i32,
219  severity: &str,
220  actor: &str,
221) -> Result<Uuid, String> {
222  {
223    let mut connection = pool.get().map_err(|e| e.to_string())?;
224    if let Some(existing) = list_recent(&mut connection, true, 200)
225      .into_iter()
226      .find(|job| populate_job_matches(job, corpus_id, service_id, severity))
227    {
228      return Ok(existing.uuid);
229    }
230  }
231  let severity = severity.to_string();
232  let params =
233    serde_json::json!({ "corpus_id": corpus_id, "service_id": service_id, "severity": severity });
234  spawn_job(pool, POPULATE_REPORT_KIND, actor, params, move |progress| {
235    let mut connection = progress.pool.get().map_err(|e| e.to_string())?;
236    progress.step(0, Some(1), &format!("aggregating the {severity} report"));
237    crate::backend::populate_scope(&mut connection, corpus_id, service_id, &severity)
238      .map_err(|e| e.to_string())?;
239    progress.step(1, Some(1), "report ready");
240    Ok(serde_json::json!({ "populated": severity }))
241  })
242}
243
244/// The job `kind` for an online index rebuild.
245pub const REINDEX_KIND: &str = "reindex";
246
247/// The high-churn / append-heavy tables that benefit from periodic maintenance — index rebuilds
248/// (their indexes bloat over time) and planner-statistics refreshes (bulk imports/reruns shift
249/// their row distributions). Mirrors the autovacuum-tuned set + `docs/DB_TUNING.md`. Shared by
250/// [`spawn_reindex`] and [`spawn_analyze`].
251const MAINTENANCE_TABLES: [&str; 7] = [
252  "tasks",
253  "log_infos",
254  "log_warnings",
255  "log_errors",
256  "log_fatals",
257  "log_invalids",
258  "historical_tasks",
259];
260
261/// Spawns a background job that rebuilds the high-churn tables' indexes **online** with
262/// `REINDEX (CONCURRENTLY)` — no exclusive lock, so reads/writes continue (DB ongoing-maintenance;
263/// `docs/DB_TUNING.md`). Runs **off** the request path (rebuilds are minutes-to-hours at scale) and
264/// reports per-table progress. **Debounced:** a reindex already queued/running is reused.
265///
266/// `REINDEX ... CONCURRENTLY` forbids running inside a transaction — the job body uses a fresh
267/// pooled connection in autocommit, so this holds.
268pub fn spawn_reindex(pool: DbPool, actor: &str) -> Result<Uuid, String> {
269  {
270    let mut connection = pool.get().map_err(|e| e.to_string())?;
271    if let Some(existing) = list_recent(&mut connection, true, 200)
272      .into_iter()
273      .find(|job| job.kind == REINDEX_KIND)
274    {
275      return Ok(existing.uuid);
276    }
277  }
278  spawn_job(
279    pool,
280    REINDEX_KIND,
281    actor,
282    serde_json::json!({ "tables": MAINTENANCE_TABLES }),
283    |progress| {
284      let mut connection = progress.pool.get().map_err(|e| e.to_string())?;
285      let total = MAINTENANCE_TABLES.len() as i32;
286      for (index, table) in MAINTENANCE_TABLES.iter().enumerate() {
287        progress.step(index as i32, Some(total), &format!("reindexing {table}"));
288        // `table` is a fixed identifier (not user input), so the interpolation is injection-safe.
289        diesel::sql_query(format!("REINDEX (CONCURRENTLY) TABLE {table}"))
290          .execute(&mut connection)
291          .map_err(|e| format!("reindex {table} failed: {e}"))?;
292      }
293      progress.step(total, Some(total), "reindex complete");
294      Ok(serde_json::json!({ "reindexed": MAINTENANCE_TABLES }))
295    },
296  )
297}
298
299/// The job `kind` for a planner-statistics refresh.
300pub const ANALYZE_KIND: &str = "analyze";
301
302/// Spawns a background job that refreshes the query planner's statistics with `ANALYZE` over the
303/// high-churn tables. After a bulk import or a large rerun churns `tasks.status`, stale statistics
304/// can make the planner mis-estimate and skip the right index (e.g. the TODO leasing index,
305/// `todo_index`), so an operator can refresh them on demand instead of waiting for autovacuum's
306/// next pass (DB ongoing-maintenance; `docs/DB_TUNING.md`). `ANALYZE` is online (a brief lock per
307/// table, sampling only) and runs **off** the request path, reporting per-table progress.
308/// **Debounced:** an analyze already queued/running is reused.
309pub fn spawn_analyze(pool: DbPool, actor: &str) -> Result<Uuid, String> {
310  {
311    let mut connection = pool.get().map_err(|e| e.to_string())?;
312    if let Some(existing) = list_recent(&mut connection, true, 200)
313      .into_iter()
314      .find(|job| job.kind == ANALYZE_KIND)
315    {
316      return Ok(existing.uuid);
317    }
318  }
319  spawn_job(
320    pool,
321    ANALYZE_KIND,
322    actor,
323    serde_json::json!({ "tables": MAINTENANCE_TABLES }),
324    |progress| {
325      let mut connection = progress.pool.get().map_err(|e| e.to_string())?;
326      let total = MAINTENANCE_TABLES.len() as i32;
327      for (index, table) in MAINTENANCE_TABLES.iter().enumerate() {
328        progress.step(index as i32, Some(total), &format!("analyzing {table}"));
329        // `table` is a fixed identifier (not user input), so the interpolation is injection-safe.
330        diesel::sql_query(format!("ANALYZE {table}"))
331          .execute(&mut connection)
332          .map_err(|e| format!("analyze {table} failed: {e}"))?;
333      }
334      progress.step(total, Some(total), "analyze complete");
335      Ok(serde_json::json!({ "analyzed": MAINTENANCE_TABLES }))
336    },
337  )
338}
339
340/// Best-effort extraction of a human-readable message from a caught panic payload.
341fn panic_message(panic: &(dyn std::any::Any + Send)) -> String {
342  if let Some(s) = panic.downcast_ref::<&str>() {
343    (*s).to_string()
344  } else if let Some(s) = panic.downcast_ref::<String>() {
345    s.clone()
346  } else {
347    "unknown panic".to_string()
348  }
349}
350
351fn set_running(pool: &DbPool, job_id: i64) {
352  if let Ok(mut connection) = pool.get() {
353    let _ = diesel::update(jobs::table.filter(jobs::id.eq(job_id)))
354      .set((jobs::status.eq("running"), jobs::updated_at.eq(now)))
355      .execute(&mut connection);
356  }
357}
358
359fn finish(pool: &DbPool, job_id: i64, status: &str, message: &str, result: Option<Value>) {
360  if let Ok(mut connection) = pool.get() {
361    let _ = diesel::update(jobs::table.filter(jobs::id.eq(job_id)))
362      .set((
363        jobs::status.eq(status),
364        jobs::message.eq(message),
365        jobs::result.eq(result),
366        jobs::updated_at.eq(now),
367      ))
368      .execute(&mut connection);
369  }
370}
371
372/// The database's current wall clock as a tz-naive timestamp **in the session timezone** — i.e. the
373/// same clock `created_at`/`updated_at` are written in (Postgres `now()` stored into a `timestamp`
374/// column). Differencing a job's `updated_at` against this is therefore skew-free, unlike comparing
375/// against the *app process's* `chrono::Utc::now()` (which differs by the session offset when the
376/// DB is not on UTC). Used to compute a running job's heartbeat age (W-4 stall observability).
377/// Best-effort: returns `None` if the probe fails (a broken connection), so callers degrade to "no
378/// age" rather than reporting a bogus one.
379pub fn db_now(connection: &mut PgConnection) -> Option<NaiveDateTime> {
380  use diesel::dsl::sql;
381  use diesel::sql_types::Timestamp;
382  diesel::select(sql::<Timestamp>("LOCALTIMESTAMP"))
383    .get_result(connection)
384    .ok()
385}
386
387/// Finds a job by its external uuid handle.
388pub fn find_job(connection: &mut PgConnection, job_uuid: Uuid) -> Option<Job> {
389  jobs::table
390    .filter(jobs::uuid.eq(job_uuid))
391    .first(connection)
392    .optional()
393    .ok()
394    .flatten()
395}
396
397/// Marks any non-terminal job whose progress heartbeat has been silent for longer than
398/// `config().jobs.stale_timeout_seconds` as `interrupted` — the **runtime** complement to
399/// [`interrupt_orphans`] (which only runs at startup). It closes the W-4 zombie: a job whose body
400/// *hangs* while a long-lived frontend keeps running would otherwise sit `running` forever, leaking
401/// a thread and lying to every pending-check + the report-refresh debounce. A job that keeps
402/// `step`-ing stays live (fresh `updated_at`); only a silent one is reaped. **Self-correcting:** if
403/// a merely-slow job is reaped and later finishes, its `finish()` overwrites the status, so a
404/// generous timeout costs at most a transient `interrupted` display. Skew-free (differences against
405/// the DB clock, like [`db_now`]). Returns the count reaped; best-effort.
406///
407/// **Caveat — Rust cannot force-kill a thread:** reaping marks the DB row terminal (so accounting,
408/// pending-checks, and the refresh debounce are correct) but the hung OS thread itself runs until
409/// its body unblocks. The leak is bounded — its pooled connection is returned between `step`s and
410/// the r2d2 pool caps total connections — but the thread/stack is only reclaimed when the body
411/// returns; truly aborting a hung *blocking* job would need subprocess isolation (a SIGKILL-able
412/// child), a deliberate architecture trade. See the W-4 ledger entry.
413pub fn reap_stale(connection: &mut PgConnection) -> usize {
414  let timeout_secs = config().jobs.stale_timeout_seconds;
415  let Some(clock) = db_now(connection) else {
416    return 0; // DB clock unreadable → skip rather than reap against a bogus app clock
417  };
418  let cutoff = clock - chrono::Duration::seconds(timeout_secs);
419  diesel::update(
420    jobs::table
421      .filter(jobs::status.eq("queued").or(jobs::status.eq("running")))
422      .filter(jobs::updated_at.lt(cutoff)),
423  )
424  .set((
425    jobs::status.eq("interrupted"),
426    jobs::message.eq(format!(
427      "no progress heartbeat for over {} minutes; presumed stalled (W-4)",
428      timeout_secs / 60
429    )),
430  ))
431  .execute(connection)
432  .unwrap_or(0)
433}
434
435/// Counts jobs in terminal `status` created within the last `hours` — a current-state observability
436/// signal for "are jobs failing / stalling lately?". A **rolling window** so the gauge
437/// auto-resolves (unlike an ever-growing total-failures counter, which would alert forever after
438/// one failure). Skew-free (windowed against the DB clock). Best-effort: `0` on error.
439pub fn count_recent_with_status(connection: &mut PgConnection, status: &str, hours: i64) -> usize {
440  let Some(clock) = db_now(connection) else {
441    return 0;
442  };
443  let cutoff = clock - chrono::Duration::hours(hours);
444  jobs::table
445    .filter(jobs::status.eq(status))
446    .filter(jobs::created_at.gt(cutoff))
447    .count()
448    .get_result::<i64>(connection)
449    .map(|count| count as usize)
450    .unwrap_or(0)
451}
452
453/// Lists recent jobs, most-recent-first, capped at `limit`. With `active_only`, returns just the
454/// **pending** (non-terminal: `queued`/`running`) jobs — the fleet-wide observability check for any
455/// background-task capability. Best-effort: an error yields an empty list rather than propagating.
456pub fn list_recent(connection: &mut PgConnection, active_only: bool, limit: i64) -> Vec<Job> {
457  // Freshen first: reap heartbeat-dead jobs so neither this listing, nor the active/pending check,
458  // nor the report-refresh debounce ever counts a hung zombie as live (W-4).
459  reap_stale(connection);
460  let mut query = jobs::table
461    .order(jobs::created_at.desc())
462    .limit(limit)
463    .into_boxed();
464  if active_only {
465    query = query.filter(jobs::status.eq("queued").or(jobs::status.eq("running")));
466  }
467  query.load(connection).unwrap_or_default()
468}
469
470/// Marks any non-terminal jobs as `interrupted`; call once on frontend startup so jobs that died
471/// with a previous process are not left looking live.
472pub fn interrupt_orphans(connection: &mut PgConnection) -> usize {
473  diesel::update(jobs::table.filter(jobs::status.eq("queued").or(jobs::status.eq("running"))))
474    .set((
475      jobs::status.eq("interrupted"),
476      jobs::message.eq("interrupted by a restart"),
477    ))
478    .execute(connection)
479    .unwrap_or(0)
480}
481
482#[cfg(test)]
483mod tests {
484  use super::*;
485
486  fn job(kind: &str, params: Value) -> Job {
487    let t = chrono::NaiveDate::from_ymd_opt(2026, 6, 28)
488      .unwrap()
489      .and_hms_opt(0, 0, 0)
490      .unwrap();
491    Job {
492      id: 1,
493      uuid: Uuid::nil(),
494      kind: kind.to_string(),
495      status: "running".to_string(),
496      progress_current: 0,
497      progress_total: None,
498      message: String::new(),
499      actor: "test".to_string(),
500      params,
501      result: None,
502      created_at: t,
503      updated_at: t,
504    }
505  }
506
507  // The populate-job debounce keys on the (corpus_id, service_id, severity) it stored in `params` —
508  // and the ids are JSON numbers (i64) compared against the i32 args, the easy-to-break part.
509  #[test]
510  fn populate_job_matches_only_its_own_scope() {
511    let p = serde_json::json!({ "corpus_id": 7, "service_id": 3, "severity": "info" });
512    let j = job(POPULATE_REPORT_KIND, p);
513    assert!(populate_job_matches(&j, 7, 3, "info"));
514    assert!(!populate_job_matches(&j, 7, 3, "warning")); // other severity of the same scope
515    assert!(!populate_job_matches(&j, 8, 3, "info")); // other corpus
516    assert!(!populate_job_matches(&j, 7, 4, "info")); // other service
517    // A different kind with identical params must never be mistaken for a populate job.
518    let other = job(
519      REFRESH_REPORTS_KIND,
520      serde_json::json!({ "corpus_id": 7, "service_id": 3, "severity": "info" }),
521    );
522    assert!(!populate_job_matches(&other, 7, 3, "info"));
523  }
524}
cortex/jobs.rs

cortex/
jobs.rs
