feat(job-registry): add the job engine

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;

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<JobRegistry>) -> 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<JobRegistry>) {
+    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<JobEntry>) {
+    // 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<JobOutcome, JoinError>` returned by the spawned
+/// handler into `(JobOutcome, Option<ErrCause>)`. `cause` is `None`
+/// on Ok, `Some(_)` on Err.
+fn translate_join(
+    res: Result<JobOutcome, tokio::task::JoinError>,
+) -> (JobOutcome, Option<ErrCause>) {
+    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::<String>() {
+                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<ErrCause>, 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<AtomicU64>,
+        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);
+    }
+}

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_<NAME>_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;
+}

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<JobRegistry>` 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};

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<JobEntry>`
+/// 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<dyn JobHandler>,
+    pub(super) interval: Duration,
+    pub(super) timeout: Option<Duration>,
+    /// 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<JobState>,
+}
+
+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<Instant>,
+    /// Wall-clock time + outcome of the most recent completed run.
+    /// `None` until the first run finishes.
+    pub last_outcome: Option<(DateTime<Utc>, 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<Utc>,
+}
+
+/// In-memory job registry. `Arc<JobRegistry>` lives on `AppState`;
+/// native services `register()` during DI wiring.
+pub struct JobRegistry {
+    entries: RwLock<HashMap<String, Arc<JobEntry>>>,
+}
+
+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<dyn JobHandler>,
+        interval: Duration,
+        timeout: Option<Duration>,
+    ) -> 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<Utc>)> {
+        let guard = self.entries.read().await;
+        let mut earliest: Option<(String, DateTime<Utc>)> = 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<JobEntry>` so
+    /// callers can hold across `await` points without pinning the
+    /// outer read lock.
+    pub async fn get(&self, name: &str) -> Option<Arc<JobEntry>> {
+        let guard = self.entries.read().await;
+        guard.get(name).cloned()
+    }
+
+    /// Snapshot every registered job (used by the admin listing
+    /// endpoint). Returns owned `(name, Arc<JobEntry>)` pairs to
+    /// avoid pinning the outer lock through the HTTP response
+    /// serialisation.
+    pub async fn snapshot_all(&self) -> Vec<(String, Arc<JobEntry>)> {
+        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<dyn JobHandler> {
+        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);
+    }
+}

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");
+    }
+}