feat(job-registry): handle job without periodicity but with trigger

docs/plan/job-registry.md

@@ -247,7 +247,10 @@ pub struct JobRegistry {
 
 struct RegisteredJob {
     handler: Arc<dyn JobHandler>,
-    interval: Duration,
+    /// `None` = on-demand only (admin trigger + programmatic
+    /// `registry.trigger(name)`), never fires periodically.
+    /// `Some(dur)` = fires every `dur` AND admin-triggerable.
+    interval: Option<Duration>,
     timeout: Option<Duration>,
     /// Single-permit gate that enforces the "one in-flight run per
     /// `job_name`" invariant (see Exclusivity above). A tick that
@@ -258,7 +261,9 @@ struct RegisteredJob {
     /// `running_for_ms` in the skip warning.
     current_run_start: Arc<parking_lot::Mutex<Option<Instant>>>,
     last_outcome: Option<(chrono::DateTime<Utc>, JobOutcome)>,
-    next_run_at: chrono::DateTime<Utc>,
+    /// Only populated for periodic jobs (`interval = Some(_)`). None
+    /// for on-demand-only jobs — `pick_next` skips them.
+    next_run_at: Option<chrono::DateTime<Utc>>,
 }
 ```
 
@@ -266,13 +271,68 @@ struct RegisteredJob {
 themselves during DI:
 
 ```rust
+// Scheduled: fires every N hours AND admin-triggerable.
 registry.register(
     Arc::clone(&trash_cleanup) as Arc<dyn JobHandler>,
-    Duration::from_secs(interval_hours * 3600),
+    Some(Duration::from_secs(interval_hours * 3600)),
     None, // no timeout
 );
+
+// On-demand only: no periodic tick, but the job is still catalogued
+// so the admin endpoint can trigger it uniformly and callers get the
+// same panic-containment + exclusivity guarantees. Used by dedup GC
+// (piggybacks on trash cleanup for its main work; admin trigger for
+// operator-driven runs).
+registry.register(
+    Arc::clone(&dedup_service) as Arc<dyn JobHandler>,
+    None,     // interval — no periodic tick
+    None,     // timeout
+);
 ```
 
+**Interval semantics.**
+- `Some(dur)` — supervisor fires the job every `dur`. Also admin-triggerable.
+- `None` — supervisor never fires the job. Admin-triggerable only. Dispatch still routes through the same `JobRegistry::trigger(name)` path so the job gets the same panic-containment, timeout, exclusivity, and log-line treatment as scheduled ones.
+
+### Manual dispatch — `JobRegistry::trigger(name)`
+
+```rust
+pub async fn trigger(&self, name: &str) -> Option<JobOutcome>;
+```
+
+The single entry point for running a registered job outside the
+scheduler's tick loop. Called by:
+- The admin endpoint (`POST /api/admin/internal/trigger-job/{name}`).
+- Any service that wants a scheduler-uniform dispatch of a peer job
+  (e.g. an inline call from trash cleanup to `trigger("dedup_gc")`,
+  if we later route the piggyback through the registry).
+
+Returns `None` when the name doesn't exist. Returns `Some(JobOutcome)`
+otherwise — even when exclusivity kicks the trigger out (that maps
+to `Ok { count: 0, extra: {"skipped": "already_running"} }`, not
+`None`).
+
+### Design boundary — registry is a catalog, not an event system
+
+Because a job can be triggered from multiple sources (scheduler,
+admin, another service), the registry visually resembles an event
+system. It is not. The distinction matters so we don't accidentally
+extend it into one.
+
+- **Registry:** *"operator or scheduler wants to run this SPECIFIC
+  named job right now."* Imperative. Single handler per name. Direct
+  dispatch. No subscription API.
+- **Event system:** *"when SOMETHING happens, notify anyone
+  interested."* Reactive. Multiple listeners per event type.
+  Publish + subscribe API. Fan-out semantics.
+
+Event-reactive work in OxiCloud goes through the existing lifecycle
+hooks — `FileLifecycleHook`, `BlobLifecycleHook`,
+`UserLifecycleHook`. Those already support multi-subscription and
+event-typed dispatch. Never add subscription machinery to
+`JobRegistry`; if a "when job A finishes, do B" case appears,
+publish a `JobCompleted` lifecycle event and let a hook subscribe.
+
 ### Engine loop
 
 ```rust

src/common/di.rs

@@ -886,7 +886,7 @@ impl AppServiceFactory {
         let interval = cleanup_service.interval();
         if let Err(e) = core
             .job_registry
-            .register(cleanup_service.clone(), interval, None)
+            .register(cleanup_service.clone(), Some(interval), None)
             .await
         {
             // Duplicate registration is the only failure mode today and
@@ -1146,7 +1146,7 @@ impl AppServiceFactory {
         );
         if let Err(e) = core
             .job_registry
-            .register(service.clone(), interval, None)
+            .register(service.clone(), Some(interval), None)
             .await
         {
             tracing::error!("Failed to register storage_reconcile job: {e}");

src/infrastructure/scheduler/engine.rs

@@ -61,6 +61,9 @@ async fn run(registry: Arc<JobRegistry>) {
     );
 
     loop {
+        // `pick_next` only returns scheduled jobs (interval = Some);
+        // on-demand jobs never appear here and are only reachable
+        // through `JobRegistry::trigger`.
         let Some((name, next_at)) = registry.pick_next().await else {
             tokio::time::sleep(IDLE_POLL).await;
             continue;
@@ -88,19 +91,28 @@ async fn run(registry: Arc<JobRegistry>) {
             continue;
         };
 
-        dispatch(&name, entry).await;
+        // Fire and forget from the supervisor's perspective — we
+        // don't care about the outcome, `dispatch` records it on the
+        // entry and emits the log line itself.
+        let _ = dispatch(&name, entry).await;
     }
 }
 
-/// Dispatch a single tick for `name`. Handles:
+/// Dispatch a single run of `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.
+/// - recording `last_outcome` + advancing `next_run_at` on completion,
+/// - emitting the uniform `oxicloud::scheduler::job.run` log line.
+///
+/// Returns the [`JobOutcome`] the run produced. The scheduler loop
+/// discards this (records-only-via-side-effect); admin/programmatic
+/// callers via [`JobRegistry::trigger`](super::registry::JobRegistry::trigger)
+/// surface it to the caller.
 ///
 /// Non-panicking; every failure path resolves to a `JobOutcome::Err`
 /// with a `cause` log field.
-async fn dispatch(name: &str, entry: Arc<JobEntry>) {
+pub(super) async fn dispatch(name: &str, entry: Arc<JobEntry>) -> JobOutcome {
     // 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.
@@ -116,17 +128,26 @@ async fn dispatch(name: &str, entry: Arc<JobEntry>) {
                     .map(|t| t.elapsed().as_millis())
                     .unwrap_or(0)
             };
+            // On-demand jobs have `interval = None`; log 0 rather than
+            // fabricate one. Operators reading this line for a scheduled
+            // job compare `interval_ms` vs `running_for_ms`; the same
+            // line for an on-demand job just tells them a concurrent
+            // trigger raced an in-flight run.
+            let interval_ms = entry.interval.map(|d| d.as_millis()).unwrap_or(0);
             tracing::warn!(
                 target: "oxicloud::scheduler",
                 event = "job.tick_skipped",
                 job = %name,
-                interval_ms = entry.interval.as_millis(),
+                interval_ms = interval_ms,
                 running_for_ms = running_for_ms,
                 "{} still running past its interval — tick skipped",
                 name,
             );
             advance_next_run(&entry);
-            return;
+            return JobOutcome::ok_with(
+                0,
+                serde_json::json!({ "skipped": "already_running" }),
+            );
         }
     };
 
@@ -169,14 +190,15 @@ async fn dispatch(name: &str, entry: Arc<JobEntry>) {
         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));
+        // Only scheduled jobs advance next_run_at. On-demand jobs stay
+        // at None so `pick_next` never returns them, even after a
+        // trigger. Same rule as the skip branch — schedule advances
+        // by one interval, no backlog queueing.
+        state.next_run_at = entry.interval.map(|dur| {
+            Utc::now()
+                + chrono::Duration::from_std(dur)
+                    .unwrap_or_else(|_| chrono::Duration::seconds(0))
+        });
     }
 
     // Log line. `outcome=ok` runs are informational; `outcome=err` include
@@ -184,16 +206,20 @@ async fn dispatch(name: &str, entry: Arc<JobEntry>) {
     log_outcome(name, &outcome, cause, elapsed_ms);
 
     drop(permit);
+    outcome
 }
 
 /// 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.
+/// run-start (skip-path helper). No-op for on-demand jobs — `interval`
+/// is `None`, so `next_run_at` stays `None` and `pick_next` continues
+/// to skip them.
 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));
+    state.next_run_at = entry.interval.map(|dur| {
+        Utc::now()
+            + chrono::Duration::from_std(dur)
+                .unwrap_or_else(|_| chrono::Duration::seconds(0))
+    });
 }
 
 /// Convert the `Result<JobOutcome, JoinError>` returned by the spawned
@@ -332,7 +358,7 @@ mod tests {
 
         let registry = Arc::new(JobRegistry::new());
         registry
-            .register(handler, Duration::from_millis(100), None)
+            .register(handler, Some(Duration::from_millis(100)), None)
             .await
             .unwrap();
         let entry = registry.get("overrun").await.unwrap();
@@ -368,7 +394,7 @@ mod tests {
         registry
             .register(
                 handler,
-                Duration::from_millis(100),
+                Some(Duration::from_millis(100)),
                 Some(Duration::from_millis(50)),
             )
             .await

src/infrastructure/scheduler/registry.rs

@@ -26,7 +26,11 @@ use super::types::JobOutcome;
 /// `await` without pinning the registry's outer lock.
 pub struct JobEntry {
     pub(super) handler: Arc<dyn JobHandler>,
-    pub(super) interval: Duration,
+    /// `None` = on-demand only; the supervisor never fires this job
+    /// (`pick_next` skips it). Admin/programmatic callers reach it
+    /// via [`JobRegistry::trigger`].
+    /// `Some(dur)` = periodic; supervisor dispatches every `dur`.
+    pub(super) interval: Option<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
@@ -45,10 +49,10 @@ pub(super) struct JobState {
     /// 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>,
+    /// Wall-clock time of the next scheduled dispatch. `None` for
+    /// on-demand jobs (never fires periodically); `Some(...)` for
+    /// scheduled jobs, advanced by one interval after every tick.
+    pub next_run_at: Option<DateTime<Utc>>,
 }
 
 /// In-memory job registry. `Arc<JobRegistry>` lives on `AppState`;
@@ -69,14 +73,19 @@ impl JobRegistry {
     /// 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.
+    /// - `interval = Some(dur)` → **scheduled**. The supervisor fires
+    ///   the job every `dur`, starting `now + dur`. Registration does
+    ///   NOT fire the job immediately — callers that want an at-startup
+    ///   run should invoke the service's own initialiser once before
+    ///   registering.
+    /// - `interval = None` → **on-demand only**. The supervisor never
+    ///   fires this job. Admin endpoint (or programmatic callers) can
+    ///   still invoke it via [`JobRegistry::trigger`] — the dispatch
+    ///   goes through the same panic/timeout/exclusivity gates.
     pub async fn register(
         &self,
         handler: Arc<dyn JobHandler>,
-        interval: Duration,
+        interval: Option<Duration>,
         timeout: Option<Duration>,
     ) -> Result<(), RegisterError> {
         let name = handler.name().to_string();
@@ -84,9 +93,11 @@ impl JobRegistry {
         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 next_run_at = interval.map(|dur| {
+            Utc::now()
+                + chrono::Duration::from_std(dur)
+                    .unwrap_or_else(|_| chrono::Duration::seconds(0))
+        });
         let entry = Arc::new(JobEntry {
             handler,
             interval,
@@ -102,18 +113,24 @@ impl JobRegistry {
         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.
+    /// Return the name and next-due timestamp of the earliest-firing
+    /// **scheduled** job, or `None` if no scheduled jobs are registered.
+    /// On-demand jobs (registered with `interval = None`) are invisible
+    /// to `pick_next` — they only run when reached via
+    /// [`Self::trigger`]. 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
+            let Some(next_at) = entry
                 .state
                 .lock()
                 .expect("JobState mutex poisoned")
-                .next_run_at;
+                .next_run_at
+            else {
+                continue; // on-demand only — never picked
+            };
             match &earliest {
                 None => earliest = Some((name.clone(), next_at)),
                 Some((_, current)) if next_at < *current => {
@@ -151,6 +168,31 @@ impl JobRegistry {
     pub async fn is_empty(&self) -> bool {
         self.entries.read().await.is_empty()
     }
+
+    /// Manual dispatch — the single entry point for running a
+    /// registered job outside the scheduler's tick loop. Called by:
+    ///
+    /// - The admin endpoint `POST /api/admin/internal/trigger-job/{name}`.
+    /// - Any service that wants a scheduler-uniform dispatch of a
+    ///   peer job (uniform log line, exclusivity, panic containment,
+    ///   timeout enforcement).
+    ///
+    /// Returns `None` when the name isn't registered. Returns
+    /// `Some(JobOutcome)` when it is — including the case where
+    /// exclusivity denied the trigger (previous run still in flight),
+    /// which surfaces as
+    /// `Ok { count: 0, extra: { "skipped": "already_running" } }` per
+    /// the engine's dispatch protocol.
+    ///
+    /// Works for BOTH scheduled and on-demand jobs — for on-demand
+    /// jobs this is the only way they ever run.
+    pub async fn trigger(
+        self: &Arc<Self>,
+        name: &str,
+    ) -> Option<JobOutcome> {
+        let entry = self.get(name).await?;
+        Some(super::engine::dispatch(name, entry).await)
+    }
 }
 
 impl Default for JobRegistry {
@@ -193,10 +235,10 @@ mod tests {
     #[tokio::test]
     async fn register_and_pick_next() {
         let reg = JobRegistry::new();
-        reg.register(handler("job_a"), Duration::from_secs(60), None)
+        reg.register(handler("job_a"), Some(Duration::from_secs(60)), None)
             .await
             .unwrap();
-        reg.register(handler("job_b"), Duration::from_secs(10), None)
+        reg.register(handler("job_b"), Some(Duration::from_secs(10)), None)
             .await
             .unwrap();
 
@@ -208,11 +250,11 @@ mod tests {
     #[tokio::test]
     async fn duplicate_registration_rejected() {
         let reg = JobRegistry::new();
-        reg.register(handler("job_x"), Duration::from_secs(60), None)
+        reg.register(handler("job_x"), Some(Duration::from_secs(60)), None)
             .await
             .unwrap();
         let err = reg
-            .register(handler("job_x"), Duration::from_secs(60), None)
+            .register(handler("job_x"), Some(Duration::from_secs(60)), None)
             .await
             .expect_err("duplicate name must be rejected");
         assert!(matches!(err, RegisterError::DuplicateName(_)));
@@ -227,13 +269,45 @@ mod tests {
     #[tokio::test]
     async fn snapshot_all_returns_every_entry() {
         let reg = JobRegistry::new();
-        reg.register(handler("a"), Duration::from_secs(1), None)
+        reg.register(handler("a"), Some(Duration::from_secs(1)), None)
             .await
             .unwrap();
-        reg.register(handler("b"), Duration::from_secs(1), None)
+        reg.register(handler("b"), Some(Duration::from_secs(1)), None)
             .await
             .unwrap();
         let all = reg.snapshot_all().await;
         assert_eq!(all.len(), 2);
     }
+
+    #[tokio::test]
+    async fn on_demand_job_invisible_to_pick_next() {
+        let reg = JobRegistry::new();
+        // Scheduled job with a long interval.
+        reg.register(handler("scheduled"), Some(Duration::from_secs(3600)), None)
+            .await
+            .unwrap();
+        // On-demand job — supervisor must never pick it.
+        reg.register(handler("on_demand"), None, None).await.unwrap();
+
+        let (next_name, _) = reg.pick_next().await.expect("scheduled job due");
+        assert_eq!(
+            next_name, "scheduled",
+            "pick_next must ignore on-demand jobs"
+        );
+    }
+
+    #[tokio::test]
+    async fn trigger_dispatches_on_demand_job() {
+        let reg = Arc::new(JobRegistry::new());
+        reg.register(handler("gc"), None, None).await.unwrap();
+
+        let outcome = reg.trigger("gc").await.expect("job exists");
+        assert!(outcome.is_ok());
+    }
+
+    #[tokio::test]
+    async fn trigger_returns_none_for_unknown_job() {
+        let reg = Arc::new(JobRegistry::new());
+        assert!(reg.trigger("nope").await.is_none());
+    }
 }