From f5f794fde5f6fa9537978d3768aacb15b783f37e Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 27 Jul 2026 22:49:51 +0200 Subject: [PATCH] feat(job-registry): handle job without periodicity but with trigger --- docs/plan/job-registry.md | 66 +++++++++++- src/common/di.rs | 4 +- src/infrastructure/scheduler/engine.rs | 68 +++++++++---- src/infrastructure/scheduler/registry.rs | 122 ++++++++++++++++++----- 4 files changed, 210 insertions(+), 50 deletions(-) diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index c6cb1b83..0e4a2f5b 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -247,7 +247,10 @@ pub struct JobRegistry { struct RegisteredJob { handler: Arc, - 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, timeout: Option, /// 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>>, last_outcome: Option<(chrono::DateTime, JobOutcome)>, - next_run_at: chrono::DateTime, + /// Only populated for periodic jobs (`interval = Some(_)`). None + /// for on-demand-only jobs — `pick_next` skips them. + next_run_at: Option>, } ``` @@ -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, - 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, + 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; +``` + +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 diff --git a/src/common/di.rs b/src/common/di.rs index 4ed769bd..e38524a7 100644 --- a/src/common/di.rs +++ b/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}"); diff --git a/src/infrastructure/scheduler/engine.rs b/src/infrastructure/scheduler/engine.rs index 324932b7..2d035439 100644 --- a/src/infrastructure/scheduler/engine.rs +++ b/src/infrastructure/scheduler/engine.rs @@ -61,6 +61,9 @@ async fn run(registry: Arc) { ); 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) { 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) { +pub(super) async fn dispatch(name: &str, entry: Arc) -> 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) { .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) { 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) { 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` 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 diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index db427c02..eaab8d57 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/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, - 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, pub(super) timeout: Option, /// 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, 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, + /// 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>, } /// In-memory job registry. `Arc` 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, - interval: Duration, + interval: Option, timeout: Option, ) -> 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)> { let guard = self.entries.read().await; let mut earliest: Option<(String, DateTime)> = 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, + name: &str, + ) -> Option { + 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()); + } }