From 8bb62affc84730ea2dac5a3521abadfd5c3eb218 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 27 Jul 2026 22:03:05 +0200 Subject: [PATCH] feat(job-registry): add the job engine --- src/infrastructure/mod.rs | 1 + src/infrastructure/scheduler/engine.rs | 388 +++++++++++++++++++++++ src/infrastructure/scheduler/handler.rs | 70 ++++ src/infrastructure/scheduler/mod.rs | 33 ++ src/infrastructure/scheduler/registry.rs | 239 ++++++++++++++ src/infrastructure/scheduler/types.rs | 116 +++++++ 6 files changed, 847 insertions(+) create mode 100644 src/infrastructure/scheduler/engine.rs create mode 100644 src/infrastructure/scheduler/handler.rs create mode 100644 src/infrastructure/scheduler/mod.rs create mode 100644 src/infrastructure/scheduler/registry.rs create mode 100644 src/infrastructure/scheduler/types.rs diff --git a/src/infrastructure/mod.rs b/src/infrastructure/mod.rs index c6383e4d..ee93db7c 100644 --- a/src/infrastructure/mod.rs +++ b/src/infrastructure/mod.rs @@ -2,4 +2,5 @@ pub mod adapters; pub mod auth_factory; pub mod db; pub mod repositories; +pub mod scheduler; pub mod services; diff --git a/src/infrastructure/scheduler/engine.rs b/src/infrastructure/scheduler/engine.rs new file mode 100644 index 00000000..324932b7 --- /dev/null +++ b/src/infrastructure/scheduler/engine.rs @@ -0,0 +1,388 @@ +//! The scheduler supervisor loop. +//! +//! One `tokio::spawn` at startup runs [`SchedulerEngine::run`]. The +//! loop iterates: +//! +//! 1. `pick_next()` — find the job with the earliest `next_run_at`. +//! 2. Sleep until that instant. +//! 3. Dispatch: try-acquire the job's in-flight permit; if held, warn +//! and reschedule; otherwise spawn the handler, apply the timeout, +//! catch panics, record the outcome. +//! +//! Sequential dispatch is intentional. Two jobs due at the same +//! instant run one-after-the-other — the second's `pick_next` fires +//! immediately after the first's dispatch returns, with a zero-length +//! sleep. See `docs/plan/job-registry.md` Part 1 §Runtime model. + +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use chrono::Utc; +use tokio::task::JoinHandle; + +use super::registry::{JobEntry, JobRegistry}; +use super::types::{ErrCause, JobOutcome}; + +/// Public handle to the running supervisor. +/// +/// Dropping the handle does NOT cancel the loop (the spawned task +/// runs until the runtime dies). Explicit shutdown is deferred to +/// whenever graceful-shutdown lands globally — matches the shape +/// every other daemon in the codebase has today. See +/// `docs/plan/job-registry.md` Part 1 §Shutdown coordination. +pub struct SchedulerEngine { + _handle: JoinHandle<()>, +} + +impl SchedulerEngine { + /// Spawn the supervisor loop and return a handle. Callers hold + /// the returned `SchedulerEngine` on `AppState` so the task lives + /// as long as the runtime. + pub fn start(registry: Arc) -> Self { + let handle = tokio::spawn(async move { + run(registry).await; + }); + Self { _handle: handle } + } +} + +/// If the registry is empty (no jobs registered yet), sleep this long +/// before rechecking. Registration happens once at boot in the current +/// design, so this only matters as a defensive fallback — in practice +/// the loop enters this branch at most once, right before the first +/// `register()` call completes. +const IDLE_POLL: Duration = Duration::from_secs(60); + +async fn run(registry: Arc) { + tracing::info!( + target: "oxicloud::scheduler", + event = "scheduler.started", + "periodic scheduler supervisor started" + ); + + loop { + let Some((name, next_at)) = registry.pick_next().await else { + tokio::time::sleep(IDLE_POLL).await; + continue; + }; + + // Convert to `Duration`. If `next_at` is in the past (missed + // tick, e.g. very short interval and the previous dispatch + // took longer than the interval), sleep zero and dispatch + // immediately. + let now = Utc::now(); + let sleep_dur = (next_at - now) + .to_std() + .unwrap_or_else(|_| Duration::from_millis(0)); + if !sleep_dur.is_zero() { + tokio::time::sleep(sleep_dur).await; + } + + // The job's `next_run_at` might have changed since `pick_next` + // returned if a concurrent trigger fired — that's fine; the + // dispatch below re-reads via the `JobEntry` snapshot. + let Some(entry) = registry.get(&name).await else { + // Job was unregistered between pick_next and dispatch — + // unreachable in the current design (no unregister), but + // guard defensively. + continue; + }; + + dispatch(&name, entry).await; + } +} + +/// Dispatch a single tick for `name`. Handles: +/// - exclusivity: try-acquire the in-flight permit; skip + warn if held, +/// - spawning under panic containment (via `tokio::spawn` + `JoinHandle`), +/// - timeout enforcement (if `ScheduledJob.timeout` is set), +/// - recording `last_outcome` + advancing `next_run_at` on completion. +/// +/// Non-panicking; every failure path resolves to a `JobOutcome::Err` +/// with a `cause` log field. +async fn dispatch(name: &str, entry: Arc) { + // Try to acquire the single-permit gate. `try_acquire` is + // non-blocking — if held, we know the previous run is still + // executing and skip this tick. + let permit = match entry.in_flight.try_acquire() { + Ok(p) => p, + Err(_) => { + // Someone else holds the permit → previous run still in + // flight. Emit the operator-signal warning and reschedule. + let running_for_ms = { + let state = entry.state.lock().expect("JobState mutex poisoned"); + state + .current_run_start + .map(|t| t.elapsed().as_millis()) + .unwrap_or(0) + }; + tracing::warn!( + target: "oxicloud::scheduler", + event = "job.tick_skipped", + job = %name, + interval_ms = entry.interval.as_millis(), + running_for_ms = running_for_ms, + "{} still running past its interval — tick skipped", + name, + ); + advance_next_run(&entry); + return; + } + }; + + // We hold the permit. Record run-start, spawn, await, translate. + { + let mut state = entry.state.lock().expect("JobState mutex poisoned"); + state.current_run_start = Some(Instant::now()); + } + let started_wall = Utc::now(); + let start_instant = Instant::now(); + + // Spawn so panics land as `JoinError::is_panic()` instead of + // unwinding into the supervisor loop. + let handler = entry.handler.clone(); + let join = tokio::spawn(async move { handler.run().await }); + + let (outcome, cause) = match entry.timeout { + Some(dur) => match tokio::time::timeout(dur, join).await { + Ok(res) => translate_join(res), + Err(_elapsed) => { + // Timeout fired. The JoinHandle is dropped, which + // aborts the spawned task cooperatively — but abort + // is best-effort in Rust; a handler that ignores + // yield points may run to completion in the background. + // We still record timeout and release the permit. + ( + JobOutcome::Err(format!("wall-clock timeout of {:?} exceeded", dur)), + Some(ErrCause::Timeout), + ) + } + }, + None => translate_join(join.await), + }; + + let elapsed_ms = start_instant.elapsed().as_millis(); + + // Record outcome and advance the schedule. Permit drops naturally + // when `permit` goes out of scope at the end of the function. + { + let mut state = entry.state.lock().expect("JobState mutex poisoned"); + state.current_run_start = None; + state.last_outcome = Some((started_wall, outcome.clone())); + // Same rule as the skip branch — schedule advances by one + // interval, no backlog queueing. If this run took longer than + // the interval, the next `pick_next` immediately sees a past + // due time and dispatches with zero sleep, but the exclusivity + // check keeps the in-flight guarantee intact. + state.next_run_at = Utc::now() + + chrono::Duration::from_std(entry.interval) + .unwrap_or_else(|_| chrono::Duration::seconds(0)); + } + + // Log line. `outcome=ok` runs are informational; `outcome=err` include + // the diagnostic `cause` field. + log_outcome(name, &outcome, cause, elapsed_ms); + + drop(permit); +} + +/// Advance `next_run_at` by one interval without touching outcome or +/// run-start (skip-path helper). Keeps the schedule steady rather +/// than queueing missed ticks. +fn advance_next_run(entry: &JobEntry) { + let mut state = entry.state.lock().expect("JobState mutex poisoned"); + state.next_run_at = Utc::now() + + chrono::Duration::from_std(entry.interval) + .unwrap_or_else(|_| chrono::Duration::seconds(0)); +} + +/// Convert the `Result` returned by the spawned +/// handler into `(JobOutcome, Option)`. `cause` is `None` +/// on Ok, `Some(_)` on Err. +fn translate_join( + res: Result, +) -> (JobOutcome, Option) { + match res { + Ok(outcome) => { + let cause = if outcome.is_ok() { + None + } else { + Some(ErrCause::Handler) + }; + (outcome, cause) + } + Err(join_err) if join_err.is_panic() => { + let payload = join_err.into_panic(); + let msg = if let Some(s) = payload.downcast_ref::<&'static str>() { + (*s).to_string() + } else if let Some(s) = payload.downcast_ref::() { + s.clone() + } else { + "unknown panic payload".to_string() + }; + ( + JobOutcome::Err(format!("handler panicked: {msg}")), + Some(ErrCause::Panicked), + ) + } + Err(join_err) => ( + JobOutcome::Err(format!("task cancelled: {join_err}")), + Some(ErrCause::Handler), + ), + } +} + +/// Emit the uniform `oxicloud::scheduler` log line for a completed run. +/// Distinct Ok/Err branches so the tracing macros pick up the fields at +/// compile time — `tracing` doesn't expand conditional field lists. +fn log_outcome(name: &str, outcome: &JobOutcome, cause: Option, elapsed_ms: u128) { + match outcome { + JobOutcome::Ok { count, extra } => { + tracing::info!( + target: "oxicloud::scheduler", + event = "job.run", + job = %name, + outcome = "ok", + count = *count, + elapsed_ms = elapsed_ms, + extra = %extra, + "job {} ran", + name, + ); + } + JobOutcome::Err(msg) => { + tracing::warn!( + target: "oxicloud::scheduler", + event = "job.run", + job = %name, + outcome = "err", + cause = %cause.unwrap_or(ErrCause::Handler), + elapsed_ms = elapsed_ms, + error = %msg, + "job {} failed", + name, + ); + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::infrastructure::scheduler::handler::JobHandler; + use async_trait::async_trait; + use std::sync::atomic::{AtomicU64, Ordering}; + + struct CountingHandler { + name: String, + calls: Arc, + sleep: Duration, + } + + #[async_trait] + impl JobHandler for CountingHandler { + fn name(&self) -> &str { + &self.name + } + async fn run(&self) -> JobOutcome { + self.calls.fetch_add(1, Ordering::SeqCst); + if !self.sleep.is_zero() { + tokio::time::sleep(self.sleep).await; + } + JobOutcome::ok(1) + } + } + + struct PanickingHandler; + + #[async_trait] + impl JobHandler for PanickingHandler { + fn name(&self) -> &str { + "panicker" + } + async fn run(&self) -> JobOutcome { + panic!("intentional test panic"); + } + } + + #[tokio::test] + async fn panic_containment_via_translate_join() { + // Directly exercise translate_join with a spawned panic — the + // supervisor loop's dispatch path uses this same helper. + let handler = Arc::new(PanickingHandler); + let join = tokio::spawn(async move { handler.run().await }); + let (outcome, cause) = translate_join(join.await); + assert!(!outcome.is_ok()); + assert_eq!(cause, Some(ErrCause::Panicked)); + if let JobOutcome::Err(msg) = outcome { + assert!(msg.contains("panicked"), "expected panic marker in: {msg}"); + } + } + + #[tokio::test] + async fn overrun_skips_second_tick() { + // Handler that sleeps 200 ms; two dispatches fired back-to-back + // should see the second skip with a `tick_skipped` warning. + let calls = Arc::new(AtomicU64::new(0)); + let handler = Arc::new(CountingHandler { + name: "overrun".to_string(), + calls: calls.clone(), + sleep: Duration::from_millis(200), + }); + + let registry = Arc::new(JobRegistry::new()); + registry + .register(handler, Duration::from_millis(100), None) + .await + .unwrap(); + let entry = registry.get("overrun").await.unwrap(); + + // Kick off dispatch 1 in the background — it holds the permit + // for ~200 ms. + let entry_bg = entry.clone(); + let bg = tokio::spawn(async move { dispatch("overrun", entry_bg).await }); + + // Give dispatch 1 time to grab the permit. + tokio::time::sleep(Duration::from_millis(50)).await; + + // Dispatch 2 should observe the permit taken and skip. + dispatch("overrun", entry.clone()).await; + + // Only dispatch 1's handler should have actually run so far. + assert_eq!(calls.load(Ordering::SeqCst), 1); + + bg.await.unwrap(); + assert_eq!(calls.load(Ordering::SeqCst), 1); + } + + #[tokio::test] + async fn timeout_records_err_and_releases_permit() { + let calls = Arc::new(AtomicU64::new(0)); + let handler = Arc::new(CountingHandler { + name: "slow".to_string(), + calls: calls.clone(), + sleep: Duration::from_millis(500), + }); + + let registry = Arc::new(JobRegistry::new()); + registry + .register( + handler, + Duration::from_millis(100), + Some(Duration::from_millis(50)), + ) + .await + .unwrap(); + let entry = registry.get("slow").await.unwrap(); + + dispatch("slow", entry.clone()).await; + + // The timeout fired; last_outcome must be Err. + let state = entry.state.lock().unwrap(); + let (_, outcome) = state.last_outcome.as_ref().expect("outcome recorded"); + assert!(!outcome.is_ok(), "expected timeout-Err, got {outcome:?}"); + + // Permit released — another dispatch could acquire it. + assert_eq!(entry.in_flight.available_permits(), 1); + } +} diff --git a/src/infrastructure/scheduler/handler.rs b/src/infrastructure/scheduler/handler.rs new file mode 100644 index 00000000..8556e257 --- /dev/null +++ b/src/infrastructure/scheduler/handler.rs @@ -0,0 +1,70 @@ +//! The implementor-facing contract for a Part 1 (periodic) job. +//! +//! Everything a native service needs to write to plug into the periodic +//! scheduler is on this page. See `docs/plan/job-registry.md` Part 1 +//! for the design rationale and migration criterion (the "operator +//! trigger" question — if an operator would never `POST /trigger-job` +//! for this loop, it doesn't belong here; keep it as a core worker). + +use async_trait::async_trait; + +use super::types::JobOutcome; + +/// Implemented by every service that wants to run on a fixed interval +/// through the periodic scheduler. +/// +/// # Design shape +/// +/// A single method, `run()`. One tick = one call. The supervisor: +/// - fires it at the registered interval, +/// - catches panics (bad handlers crash their own run, not the scheduler), +/// - enforces the configured wall-clock timeout (if any), +/// - enforces exclusivity — a second tick that fires while a previous +/// run is still executing is **skipped, not queued**, with a +/// `job.tick_skipped` warning emitted (the operator signal that the +/// job is chronically slower than its cadence). +/// +/// Implementors write the body of `run()`. Everything else — logging, +/// timing, panic containment, exclusivity — is the supervisor's job. +/// +/// # `name()` guidance +/// +/// Return a stable, unique snake_case identifier. Log lines +/// (`job = %name`), admin listing, admin trigger URLs +/// (`POST /api/admin/internal/trigger-job/{name}`) and env vars +/// (`OXICLOUD_JOB__INTERVAL_HOURS`) all key on this. Renaming +/// after release is a breaking change to operator scripts and log +/// dashboards. +/// +/// # `run()` guidance +/// +/// Return [`JobOutcome::Ok`] with a `count` scalar the operator finds +/// meaningful (rows swept, blobs GC'd, bytes reclaimed) plus optional +/// `extra` JSON. Return [`JobOutcome::Err`] on failure — the +/// supervisor logs it under `outcome=err, cause=handler` and moves +/// on; the next tick fires normally. +/// +/// **Do not** catch panics inside `run()` — the supervisor does it, +/// and hiding one loses the `cause=panicked` diagnostic signal. +/// +/// **Do not** call `tokio::time::sleep` for long durations inside +/// `run()` if you have a `timeout` configured — the timeout fires +/// mid-sleep and kills the run with `cause=timeout`. Use short polling +/// intervals or restructure the work. +/// +/// # Reference implementation +/// +/// See `TrashCleanupService::run` (once migrated) as the canonical +/// example: reads its own configuration, runs bounded work, returns +/// a count. Everything else is boilerplate the scheduler owns. +#[async_trait] +pub trait JobHandler: Send + Sync { + /// Stable, unique snake_case identifier. Must be unique across + /// the process; the registry rejects duplicate registration. + fn name(&self) -> &str; + + /// One execution. Called at the registered interval and (optionally) + /// on admin trigger. See trait-level docs for guidance on when to + /// return Ok vs Err. + async fn run(&self) -> JobOutcome; +} diff --git a/src/infrastructure/scheduler/mod.rs b/src/infrastructure/scheduler/mod.rs new file mode 100644 index 00000000..36d62e79 --- /dev/null +++ b/src/infrastructure/scheduler/mod.rs @@ -0,0 +1,33 @@ +//! Periodic job scheduler (Part 1 of the job-registry plan). +//! +//! In-memory registration + one-supervisor-task dispatch. Fires +//! registered [`JobHandler`] implementations at their configured +//! intervals with panic containment, timeout enforcement, and +//! same-name exclusivity. +//! +//! # For future implementors +//! +//! - **You write a `JobHandler`.** Implement [`JobHandler::name`] + +//! [`JobHandler::run`] on your service. Nothing else. See +//! [`handler`] for the guidance doc-comment. +//! - **DI wires the registration.** In `common/di.rs` (or wherever the +//! composition root lives), build an `Arc` once, register +//! every service that opts in, then call [`SchedulerEngine::start`]. +//! - **Should this loop actually be a scheduler job?** See the migration +//! criterion in `docs/plan/job-registry.md` — the primary question +//! is "would an operator plausibly trigger this manually?". Continuous +//! drains and event-reactive workers stay as their own loops. +//! +//! Part 2 (recoverable-run engine, DB-backed cursor + resume) is +//! designed but not yet implemented. When it lands it will slot in +//! as a sibling module without changing anything here. + +mod engine; +mod handler; +mod registry; +mod types; + +pub use engine::SchedulerEngine; +pub use handler::JobHandler; +pub use registry::{JobEntry, JobRegistry, RegisterError}; +pub use types::{ErrCause, JobOutcome}; diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs new file mode 100644 index 00000000..db427c02 --- /dev/null +++ b/src/infrastructure/scheduler/registry.rs @@ -0,0 +1,239 @@ +//! In-memory registry of periodic jobs. +//! +//! The [`JobRegistry`] owns a map `name → JobEntry`. Native services +//! `register()` themselves during DI; the [`SchedulerEngine`](super::engine::SchedulerEngine) +//! iterates this map on every tick to pick the next-due job. +//! +//! Per-job state (in-flight semaphore, last outcome, next-run time) +//! lives inside each [`JobEntry`] behind a short-lived `std::sync::Mutex`. +//! The outer map uses a `tokio::sync::RwLock` so `pick_next` and +//! `snapshot` don't block one another and so future dynamic +//! registration (plugin manifests, admin UI) can acquire a write +//! lock without racing readers. + +use std::collections::HashMap; +use std::sync::{Arc, Mutex}; +use std::time::{Duration, Instant}; + +use chrono::{DateTime, Utc}; +use tokio::sync::{RwLock, Semaphore}; + +use super::handler::JobHandler; +use super::types::JobOutcome; + +/// A registered job plus its runtime state. Held as `Arc` +/// inside the registry so the engine can hold a snapshot across an +/// `await` without pinning the registry's outer lock. +pub struct JobEntry { + pub(super) handler: Arc, + pub(super) interval: Duration, + pub(super) timeout: Option, + /// Single-permit gate enforcing the "one in-flight run per + /// `job_name`" invariant. A tick that finds the permit taken + /// emits `job.tick_skipped` and does not spawn. + pub(super) in_flight: Semaphore, + /// Mutable state — protected by `std::sync::Mutex` because guards + /// are only held for a few statements at a time, never across an + /// `await`. `tokio::sync::Mutex` would add overhead for no benefit. + pub(super) state: Mutex, +} + +pub(super) struct JobState { + /// Set when a run starts, cleared when it ends. Used to include + /// `running_for_ms` in the `job.tick_skipped` warning. + pub current_run_start: Option, + /// Wall-clock time + outcome of the most recent completed run. + /// `None` until the first run finishes. + pub last_outcome: Option<(DateTime, JobOutcome)>, + /// Wall-clock time of the next scheduled dispatch. Advanced by + /// one interval after every tick — both successful dispatch and + /// skipped (in-flight) tick. + pub next_run_at: DateTime, +} + +/// In-memory job registry. `Arc` lives on `AppState`; +/// native services `register()` during DI wiring. +pub struct JobRegistry { + entries: RwLock>>, +} + +impl JobRegistry { + pub fn new() -> Self { + Self { + entries: RwLock::new(HashMap::new()), + } + } + + /// Register a job. Returns an error if a job with the same name + /// is already registered — names are the primary identifier + /// everywhere (logs, admin URLs, env vars) and collisions would + /// hide bugs. + /// + /// `first_run_at` = `Utc::now() + interval` by convention — + /// registration doesn't fire the job immediately. Callers that + /// want an at-startup run should invoke the service's own + /// initialiser once before registering. + pub async fn register( + &self, + handler: Arc, + interval: Duration, + timeout: Option, + ) -> Result<(), RegisterError> { + let name = handler.name().to_string(); + let mut guard = self.entries.write().await; + if guard.contains_key(&name) { + return Err(RegisterError::DuplicateName(name)); + } + let now = Utc::now(); + let next_run_at = now + + chrono::Duration::from_std(interval).unwrap_or_else(|_| chrono::Duration::seconds(0)); + let entry = Arc::new(JobEntry { + handler, + interval, + timeout, + in_flight: Semaphore::new(1), + state: Mutex::new(JobState { + current_run_start: None, + last_outcome: None, + next_run_at, + }), + }); + guard.insert(name, entry); + Ok(()) + } + + /// Return the name and next-due timestamp of the job that fires + /// soonest, or `None` if no jobs are registered. Read-lock only — + /// safe to call frequently from the supervisor loop. + pub async fn pick_next(&self) -> Option<(String, DateTime)> { + let guard = self.entries.read().await; + let mut earliest: Option<(String, DateTime)> = None; + for (name, entry) in guard.iter() { + let next_at = entry + .state + .lock() + .expect("JobState mutex poisoned") + .next_run_at; + match &earliest { + None => earliest = Some((name.clone(), next_at)), + Some((_, current)) if next_at < *current => { + earliest = Some((name.clone(), next_at)) + } + _ => {} + } + } + earliest + } + + /// Snapshot handle to a single job. Returns `Arc` so + /// callers can hold across `await` points without pinning the + /// outer read lock. + pub async fn get(&self, name: &str) -> Option> { + let guard = self.entries.read().await; + guard.get(name).cloned() + } + + /// Snapshot every registered job (used by the admin listing + /// endpoint). Returns owned `(name, Arc)` pairs to + /// avoid pinning the outer lock through the HTTP response + /// serialisation. + pub async fn snapshot_all(&self) -> Vec<(String, Arc)> { + let guard = self.entries.read().await; + guard.iter().map(|(k, v)| (k.clone(), v.clone())).collect() + } + + /// Count of registered jobs — used for the startup log line. + pub async fn len(&self) -> usize { + self.entries.read().await.len() + } + + #[allow(dead_code)] + pub async fn is_empty(&self) -> bool { + self.entries.read().await.is_empty() + } +} + +impl Default for JobRegistry { + fn default() -> Self { + Self::new() + } +} + +#[derive(Debug, thiserror::Error)] +pub enum RegisterError { + #[error("job name already registered: {0}")] + DuplicateName(String), +} + +#[cfg(test)] +mod tests { + use super::*; + use async_trait::async_trait; + + struct DummyHandler { + name: String, + } + + #[async_trait] + impl JobHandler for DummyHandler { + fn name(&self) -> &str { + &self.name + } + async fn run(&self) -> JobOutcome { + JobOutcome::ok(0) + } + } + + fn handler(name: &str) -> Arc { + Arc::new(DummyHandler { + name: name.to_string(), + }) + } + + #[tokio::test] + async fn register_and_pick_next() { + let reg = JobRegistry::new(); + reg.register(handler("job_a"), Duration::from_secs(60), None) + .await + .unwrap(); + reg.register(handler("job_b"), Duration::from_secs(10), None) + .await + .unwrap(); + + let (next_name, _) = reg.pick_next().await.expect("expected a due job"); + // job_b has the shorter interval → earlier next_run_at. + assert_eq!(next_name, "job_b"); + } + + #[tokio::test] + async fn duplicate_registration_rejected() { + let reg = JobRegistry::new(); + reg.register(handler("job_x"), Duration::from_secs(60), None) + .await + .unwrap(); + let err = reg + .register(handler("job_x"), Duration::from_secs(60), None) + .await + .expect_err("duplicate name must be rejected"); + assert!(matches!(err, RegisterError::DuplicateName(_))); + } + + #[tokio::test] + async fn empty_registry_picks_nothing() { + let reg = JobRegistry::new(); + assert!(reg.pick_next().await.is_none()); + } + + #[tokio::test] + async fn snapshot_all_returns_every_entry() { + let reg = JobRegistry::new(); + reg.register(handler("a"), Duration::from_secs(1), None) + .await + .unwrap(); + reg.register(handler("b"), Duration::from_secs(1), None) + .await + .unwrap(); + let all = reg.snapshot_all().await; + assert_eq!(all.len(), 2); + } +} diff --git a/src/infrastructure/scheduler/types.rs b/src/infrastructure/scheduler/types.rs new file mode 100644 index 00000000..bdc9b438 --- /dev/null +++ b/src/infrastructure/scheduler/types.rs @@ -0,0 +1,116 @@ +//! Common types shared across the scheduler module. +//! +//! Nothing here talks to the DB or the async runtime — pure data +//! definitions so downstream modules (handler, registry, engine) can +//! import without dragging in transitive dependencies. See +//! `docs/plan/job-registry.md` Part 1 for the design rationale. + +use std::fmt; + +use serde::{Deserialize, Serialize}; + +/// Uniform outcome the supervisor logs and stores for every job dispatch. +/// +/// Two variants, deliberately. Distinguishing *why* a job failed +/// (handler returned Err, `tokio::time::timeout` tripped, +/// `catch_unwind` caught a panic) is a **diagnostic** concern — it +/// belongs in a `cause` tracing field the supervisor sets, not in a +/// control-flow branch every consumer of `match outcome` has to +/// think about. See `docs/plan/job-registry.md` Part 1 §JobOutcome. +/// +/// `Ok::count` is the row/record count the job reports as its primary +/// scalar (rows scanned, blobs migrated, thumbnails checked). `extra` +/// is a free-form JSON blob for job-specific fields the caller wants +/// surfaced to `oxicloud::scheduler` log lines. +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "outcome", rename_all = "snake_case")] +pub enum JobOutcome { + Ok { + count: u64, + #[serde(default, skip_serializing_if = "serde_json::Value::is_null")] + extra: serde_json::Value, + }, + Err(String), +} + +impl JobOutcome { + /// Ok with no extras — the common case for jobs that only report a count. + pub fn ok(count: u64) -> Self { + JobOutcome::Ok { + count, + extra: serde_json::Value::Null, + } + } + + /// Ok with a JSON `extra` payload. Use `serde_json::json!({...})` + /// at call sites for readability. + pub fn ok_with(count: u64, extra: serde_json::Value) -> Self { + JobOutcome::Ok { count, extra } + } + + /// Terse discriminant for logs / metrics: `"ok"` | `"err"`. + pub fn kind(&self) -> &'static str { + match self { + JobOutcome::Ok { .. } => "ok", + JobOutcome::Err(_) => "err", + } + } + + pub fn is_ok(&self) -> bool { + matches!(self, JobOutcome::Ok { .. }) + } +} + +/// Diagnostic reason the supervisor attaches to the `cause` tracing +/// field when a job's outcome is [`JobOutcome::Err`]. Never persisted +/// as a first-class column — it's a log field only. +/// +/// Handlers never construct this; the supervisor derives it from +/// which failure path fired: +/// - [`ErrCause::Handler`] — the handler returned `Err(_)` itself. +/// - [`ErrCause::Timeout`] — `tokio::time::timeout` tripped on the +/// registered `ScheduledJob.timeout` wall-clock cap. +/// - [`ErrCause::Panicked`] — `JoinHandle` returned a panic error. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ErrCause { + Handler, + Timeout, + Panicked, +} + +impl ErrCause { + /// Stable label for the `cause` tracing field. Log aggregators key + /// on these — renaming here IS a breaking change to any dashboard + /// filtering on `cause = "handler"`. + pub fn as_str(self) -> &'static str { + match self { + ErrCause::Handler => "handler", + ErrCause::Timeout => "timeout", + ErrCause::Panicked => "panicked", + } + } +} + +impl fmt::Display for ErrCause { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.write_str(self.as_str()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn joboutcome_kind_label() { + assert_eq!(JobOutcome::ok(0).kind(), "ok"); + assert_eq!(JobOutcome::Err("boom".into()).kind(), "err"); + } + + #[test] + fn errcause_labels_stable() { + assert_eq!(ErrCause::Handler.as_str(), "handler"); + assert_eq!(ErrCause::Timeout.as_str(), "timeout"); + assert_eq!(ErrCause::Panicked.as_str(), "panicked"); + } +}