diff --git a/docs/plan/job-registry.md b/docs/plan/job-registry.md index 0e4a2f5b..c5fab85c 100644 --- a/docs/plan/job-registry.md +++ b/docs/plan/job-registry.md @@ -53,8 +53,8 @@ duplicating them between parts. Not every background loop belongs in JobRegistry. The single question that decides: -> **"Would an operator plausibly `POST /trigger-job/{name}` to make -> it run right now?"** +> **"Would an operator plausibly `POST /api/admin/jobs/{name}/trigger` +> to make it run right now?"** **Yes → migrate.** The whole payoff of JobRegistry is a uniform *operator surface* — list, trigger, last-outcome, log line, config @@ -144,7 +144,26 @@ pub trait JobHandler: Send + Sync { /// extra }` on success — the count is the primary scalar the job /// reports (rows swept, ETags flushed, blobs GC'd). Return /// `Err(msg)` on failure; the supervisor logs it and continues. - async fn run(&self) -> JobOutcome; + /// + /// `args` carries per-dispatch parameters. Periodic ticks pass + /// `JobRunArgs::default()`; admin triggers can set `force: true` + /// to request acceleration semantics (e.g. dedup GC skips its + /// orphan grace window, grant cleanup uses grace = 0). Handlers + /// that don't understand a given arg silently ignore it — no + /// return-error path just because a caller set an unused flag. + async fn run(&self, args: &JobRunArgs) -> JobOutcome; +} + +/// Per-dispatch parameters. Grows over time; today it carries only +/// `force`. Kept as a struct (not `bool`) so we don't have to change +/// signatures the next time a job needs another knob. +#[derive(Debug, Clone, Default)] +pub struct JobRunArgs { + /// Request acceleration semantics. Semantics are per-job: + /// - `dedup_gc`: skip the orphan grace window (grace = 0). + /// - `grant_cleanup`: grace = 0. + /// - Others: silently ignored. + pub force: bool, } ``` @@ -294,19 +313,22 @@ registry.register( - `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)` +### Manual dispatch — `JobRegistry::trigger(name, args)` ```rust -pub async fn trigger(&self, name: &str) -> Option; +pub async fn trigger(&self, name: &str, args: &JobRunArgs) -> 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}`). +- The admin endpoint (`POST /api/admin/jobs/{name}/trigger?force=`). - Any service that wants a scheduler-uniform dispatch of a peer job - (e.g. an inline call from trash cleanup to `trigger("dedup_gc")`, + (e.g. an inline call from trash cleanup to `trigger("dedup_gc", &args)`, if we later route the piggyback through the registry). +The supervisor's periodic ticks invoke the same underlying dispatch +with `JobRunArgs::default()` — periodic runs never force. + 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 @@ -384,11 +406,11 @@ into the scheduler. 2. **Boot**: start server; expect `scheduler started, N job(s) registered`. 3. **Admin listing**: ``` - curl -s http://localhost:8086/api/admin/internal/jobs -H "Authorization: Bearer $TOKEN" + curl -s http://localhost:8086/api/admin/jobs -H "Authorization: Bearer $TOKEN" ``` returns a JSON array with each registered job, its `interval_ms`, `next_run_at`, and `last_outcome` (null until first tick). -4. **Trigger**: `POST /api/admin/internal/trigger-job/trash_cleanup` +4. **Trigger**: `POST /api/admin/jobs/trash_cleanup/trigger` invokes the handler immediately, records the outcome. 5. **Panic containment**: unit test a handler that panics; `last_outcome` records `Err(...)` with `cause = "panicked"` in the log; the scheduler @@ -665,16 +687,17 @@ into this general one. ### Admin surface (recoverable runs) -Same URL taxonomy as Part 1, extended for run identity: +Same URL taxonomy as Part 1 — resource-first, action second, all +under `/api/admin/jobs/{name}/*`. Extended for run identity: ``` -POST /api/admin/internal/trigger-job/{name} +POST /api/admin/jobs/{name}/trigger → { run_id, status } # starts or resumes; idempotent -POST /api/admin/internal/trigger-job/{name}/cancel +POST /api/admin/jobs/{name}/cancel → { run_id, status: "CancelRequested" } -GET /api/admin/internal/jobs/{name}/runs +GET /api/admin/jobs/{name}/runs → [{ run_id, status, started_at, last_progress_at, stats, ... }] -GET /api/admin/internal/jobs/{name}/runs/{id} +GET /api/admin/jobs/{name}/runs/{id} → { run_id, status, cursor_hex, stats, params, error_message, ... } ``` @@ -696,13 +719,13 @@ GET /api/admin/internal/jobs/{name}/runs/{id} ### Verification (Part 2) 1. **Compile + schema-migration idempotence.** -2. **Fresh run:** `POST /trigger-job/storage_migration` → new row with +2. **Fresh run:** `POST /api/admin/jobs/storage_migration/trigger` → new row with `status='Running'`, `cursor=NULL`. 3. **Concurrent trigger:** second `POST` while the first is running returns the SAME `run_id` (idempotent, DB unique index enforces). -4. **Cancel + resume round-trip:** `trigger-job/…/cancel` flips to +4. **Cancel + resume round-trip:** `/api/admin/jobs/…/cancel` flips to `CancelRequested`; handler polls, returns `Paused { cursor }`; - engine writes `Paused`. `POST /trigger-job/…` again resumes; cursor + engine writes `Paused`. `POST /api/admin/jobs/…/trigger` again resumes; cursor picks up where left off; `stats.count` continues accumulating. 5. **Crash recovery:** stop the server mid-run; restart; boot sweep flips the row to `Paused` with `error_message = 'server restart mid-run'`; @@ -721,16 +744,46 @@ GET /api/admin/internal/jobs/{name}/runs/{id} ### Admin URL taxonomy -All under `/api/admin/internal/*`, gated by the existing -`OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS` env var — reuses the same -admin-guard middleware and the same "disabled → 404" contract as -today's per-service triggers. +All scheduler endpoints live on the **production admin surface**: +`/api/admin/jobs/*`. Always on, audit-logged, no feature-flag gate — +these are the operational levers you actually want ops to reach in +prod. See `project_admin_url_taxonomy` for the `/admin` vs +`/admin/internal` split we're honouring here. -**Existing per-service shims** (`trigger-sweep`, `trigger-gc`, -`trigger-grant-cleanup`) stay as thin forwards to `trigger-job/{name}` -during migration so the existing Hurl suites keep working. -Deprecation surfaces via a `Deprecation: true` response header -operators can grep for. +**Resource-first URL taxonomy** for every scheduler-owned endpoint: + +``` +GET /api/admin/jobs # list all +POST /api/admin/jobs/{name}/trigger # one dispatch (Part 1 + 2) +POST /api/admin/jobs/{name}/cancel # cooperative pause (Part 2) +GET /api/admin/jobs/{name}/runs # run history (Part 2) +GET /api/admin/jobs/{name}/runs/{id} # single run detail (Part 2) +``` + +`{name}` is the stable `JobHandler::name()` identifier. `trigger` +accepts an optional `?force=` query param that maps to +`JobRunArgs.force`. + +**Audit logging.** Every `POST` to `/api/admin/jobs/*` emits a +`target: "audit"` line before invoking the registry — bulk-effect +mutations belong on the audit stream. Success/failure outcome fires +its own `oxicloud::scheduler` line via the existing supervisor path. + +**Legacy shim retirement** (Stage 2 — follow-up PR after this one): + +The three existing internal endpoints map 1:1 to the new surface: + +| Legacy | Replacement | +|---|---| +| `POST /admin/internal/trigger-sweep` | `POST /admin/jobs/storage_reconcile/trigger` | +| `POST /admin/internal/trigger-gc?force=X` | `POST /admin/jobs/dedup_gc/trigger?force=X` | +| `POST /admin/internal/trigger-grant-cleanup?force=X` | `POST /admin/jobs/grant_cleanup/trigger?force=X` | + +Rewritten as thin forwards to the new endpoints with a `Deprecation: +true` response header while Hurl suites migrate to the new paths. Once +all callers cut over, the shims are deleted AND the +`OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS` env var is removed — its +sole purpose was gating those shims. ### Config surface — env vars diff --git a/src/application/services/storage_usage_service.rs b/src/application/services/storage_usage_service.rs index 35b1d3ec..1e23f34f 100644 --- a/src/application/services/storage_usage_service.rs +++ b/src/application/services/storage_usage_service.rs @@ -578,7 +578,7 @@ impl StorageUsageService { pub const STORAGE_RECONCILE_JOB_NAME: &str = "storage_reconcile"; -use crate::infrastructure::scheduler::{JobHandler, JobOutcome}; +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRunArgs}; use async_trait::async_trait; #[async_trait] @@ -601,7 +601,11 @@ impl JobHandler for StorageUsageService { /// Failure of one sub-sweep short-circuits the tick to `Err`; /// operators see `outcome=err, cause=handler` in the scheduler /// log and the individual sweep's own `error!` line above it. - async fn run(&self) -> JobOutcome { + /// + /// `args.force` is ignored — reconciliation is idempotent and has + /// no acceleration semantics; every run does the same set-based + /// UPDATE regardless. + async fn run(&self, _args: &JobRunArgs) -> JobOutcome { let drives = match self.update_all_drives_storage_usage().await { Ok(n) => n, Err(e) => return JobOutcome::Err(format!("drive reconciliation failed: {e}")), diff --git a/src/common/di.rs b/src/common/di.rs index fd1db87e..d9e901af 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1281,7 +1281,7 @@ impl AppServiceFactory { // sweep already runs GC as its tail step, so a periodic dedup // schedule would double the work. Registering with `interval = // None` keeps it admin-triggerable through the uniform scheduler - // surface (`POST /api/admin/internal/trigger-job/dedup_gc`). + // surface (`POST /api/admin/jobs/dedup_gc/trigger`). if let Err(e) = core .job_registry .register( diff --git a/src/infrastructure/scheduler/engine.rs b/src/infrastructure/scheduler/engine.rs index d6c05a84..54f6dc0a 100644 --- a/src/infrastructure/scheduler/engine.rs +++ b/src/infrastructure/scheduler/engine.rs @@ -21,7 +21,7 @@ use chrono::Utc; use tokio::task::JoinHandle; use super::registry::{JobEntry, JobRegistry}; -use super::types::{ErrCause, JobOutcome}; +use super::types::{ErrCause, JobOutcome, JobRunArgs}; /// Public handle to the running supervisor. /// @@ -93,8 +93,9 @@ async fn run(registry: Arc) { // 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; + // entry and emits the log line itself. Periodic ticks never + // force — that's an admin-trigger-only affordance. + let _ = dispatch(&name, entry, &JobRunArgs::default()).await; } } @@ -112,7 +113,15 @@ async fn run(registry: Arc) { /// /// Non-panicking; every failure path resolves to a `JobOutcome::Err` /// with a `cause` log field. -pub(super) async fn dispatch(name: &str, entry: Arc) -> JobOutcome { +/// +/// `args` is passed through to `JobHandler::run`. The supervisor's +/// periodic ticks pass `JobRunArgs::default()`; the admin trigger +/// endpoint forwards parsed query params such as `?force=true`. +pub(super) async fn dispatch( + name: &str, + entry: Arc, + args: &JobRunArgs, +) -> 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. @@ -157,9 +166,11 @@ pub(super) async fn dispatch(name: &str, entry: Arc) -> JobOutcome { let start_instant = Instant::now(); // Spawn so panics land as `JoinError::is_panic()` instead of - // unwinding into the supervisor loop. + // unwinding into the supervisor loop. Args cloned into the spawn + // scope so the borrow doesn't outlive the caller. let handler = entry.handler.clone(); - let join = tokio::spawn(async move { handler.run().await }); + let args_owned = args.clone(); + let join = tokio::spawn(async move { handler.run(&args_owned).await }); let (outcome, cause) = match entry.timeout { Some(dur) => match tokio::time::timeout(dur, join).await { @@ -305,7 +316,7 @@ mod tests { fn name(&self) -> &str { &self.name } - async fn run(&self) -> JobOutcome { + async fn run(&self, _args: &JobRunArgs) -> JobOutcome { self.calls.fetch_add(1, Ordering::SeqCst); if !self.sleep.is_zero() { tokio::time::sleep(self.sleep).await; @@ -321,7 +332,7 @@ mod tests { fn name(&self) -> &str { "panicker" } - async fn run(&self) -> JobOutcome { + async fn run(&self, _args: &JobRunArgs) -> JobOutcome { panic!("intentional test panic"); } } @@ -331,7 +342,7 @@ mod tests { // 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 join = tokio::spawn(async move { handler.run(&JobRunArgs::default()).await }); let (outcome, cause) = translate_join(join.await); assert!(!outcome.is_ok()); assert_eq!(cause, Some(ErrCause::Panicked)); @@ -361,13 +372,15 @@ mod tests { // 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 }); + let bg = tokio::spawn(async move { + dispatch("overrun", entry_bg, &JobRunArgs::default()).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; + dispatch("overrun", entry.clone(), &JobRunArgs::default()).await; // Only dispatch 1's handler should have actually run so far. assert_eq!(calls.load(Ordering::SeqCst), 1); @@ -396,7 +409,7 @@ mod tests { .unwrap(); let entry = registry.get("slow").await.unwrap(); - dispatch("slow", entry.clone()).await; + dispatch("slow", entry.clone(), &JobRunArgs::default()).await; // The timeout fired; last_outcome must be Err. let state = entry.state.lock().unwrap(); diff --git a/src/infrastructure/scheduler/handler.rs b/src/infrastructure/scheduler/handler.rs index 8556e257..c90e8087 100644 --- a/src/infrastructure/scheduler/handler.rs +++ b/src/infrastructure/scheduler/handler.rs @@ -3,12 +3,13 @@ //! 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). +//! trigger" question — if an operator would never +//! `POST /api/admin/jobs/{name}/trigger` for this loop, it doesn't +//! belong here; keep it as a core worker). use async_trait::async_trait; -use super::types::JobOutcome; +use super::types::{JobOutcome, JobRunArgs}; /// Implemented by every service that wants to run on a fixed interval /// through the periodic scheduler. @@ -31,7 +32,7 @@ use super::types::JobOutcome; /// /// 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 +/// (`POST /api/admin/jobs/{name}/trigger`) and env vars /// (`OXICLOUD_JOB__INTERVAL_HOURS`) all key on this. Renaming /// after release is a breaking change to operator scripts and log /// dashboards. @@ -64,7 +65,15 @@ pub trait JobHandler: Send + Sync { 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; + /// on admin trigger. + /// + /// `args` carries per-dispatch parameters (`force: bool` today). + /// Periodic ticks pass [`JobRunArgs::default()`]; admin triggers + /// forward query params such as `?force=true`. Handlers that don't + /// understand a given arg silently ignore it — the arg exists to + /// give per-job acceleration semantics without spreading per-job + /// knowledge into every caller. + /// + /// See trait-level docs for guidance on when to return Ok vs Err. + async fn run(&self, args: &JobRunArgs) -> JobOutcome; } diff --git a/src/infrastructure/scheduler/mod.rs b/src/infrastructure/scheduler/mod.rs index 36d62e79..6eb449cc 100644 --- a/src/infrastructure/scheduler/mod.rs +++ b/src/infrastructure/scheduler/mod.rs @@ -29,5 +29,5 @@ mod types; pub use engine::SchedulerEngine; pub use handler::JobHandler; -pub use registry::{JobEntry, JobRegistry, RegisterError}; -pub use types::{ErrCause, JobOutcome}; +pub use registry::{JobEntry, JobRegistry, JobSummary, RegisterError}; +pub use types::{ErrCause, JobOutcome, JobRunArgs}; diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index 4e8fe41d..96b66c7c 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/src/infrastructure/scheduler/registry.rs @@ -16,10 +16,11 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use chrono::{DateTime, Utc}; +use serde::Serialize; use tokio::sync::{RwLock, Semaphore}; use super::handler::JobHandler; -use super::types::JobOutcome; +use super::types::{JobOutcome, JobRunArgs}; /// A registered job plus its runtime state. Held as `Arc` /// inside the registry so the engine can hold a snapshot across an @@ -158,6 +159,32 @@ impl JobRegistry { guard.iter().map(|(k, v)| (k.clone(), v.clone())).collect() } + /// Serialisable snapshot for `GET /api/admin/jobs`. Each entry + /// captures the operator-visible state: interval (null for on- + /// demand), next scheduled dispatch (null for on-demand), when + /// the last run started, and its outcome. + pub async fn snapshot(&self) -> Vec { + let entries = self.snapshot_all().await; + entries + .into_iter() + .map(|(name, entry)| { + let state = entry.state.lock().expect("JobState mutex poisoned"); + let (last_run_at, last_outcome) = match &state.last_outcome { + Some((at, outcome)) => (Some(*at), Some(outcome.clone())), + None => (None, None), + }; + JobSummary { + name, + interval_ms: entry.interval.map(|d| d.as_millis() as u64), + next_run_at: state.next_run_at, + last_run_at, + last_outcome, + running: state.current_run_start.is_some(), + } + }) + .collect() + } + /// Count of registered jobs — used for the startup log line. pub async fn len(&self) -> usize { self.entries.read().await.len() @@ -171,7 +198,7 @@ impl JobRegistry { /// 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}`. + /// - The admin endpoint `POST /api/admin/jobs/{name}/trigger`. /// - Any service that wants a scheduler-uniform dispatch of a /// peer job (uniform log line, exclusivity, panic containment, /// timeout enforcement). @@ -185,9 +212,17 @@ impl JobRegistry { /// /// 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 { + /// + /// `args` is forwarded to `JobHandler::run`. Admin trigger routes + /// use `JobRunArgs { force: query.force }`; programmatic callers + /// that just want a plain run pass `JobRunArgs::default()`. + pub async fn trigger( + self: &Arc, + name: &str, + args: &JobRunArgs, + ) -> Option { let entry = self.get(name).await?; - Some(super::engine::dispatch(name, entry).await) + Some(super::engine::dispatch(name, entry, args).await) } } @@ -203,6 +238,29 @@ pub enum RegisterError { DuplicateName(String), } +/// Per-job row in the `GET /api/admin/jobs` response. +/// +/// - `interval_ms` — periodic cadence; `null` for on-demand jobs. +/// - `next_run_at` — next scheduled dispatch; `null` for on-demand. +/// - `last_run_at` / `last_outcome` — most recent completed run; +/// `null` until the first run finishes. +/// - `running` — true iff the in-flight permit is currently held +/// (either the supervisor tick is in progress or an admin trigger +/// raced in). +#[derive(Debug, Clone, Serialize)] +pub struct JobSummary { + pub name: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub interval_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub next_run_at: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_run_at: Option>, + #[serde(skip_serializing_if = "Option::is_none")] + pub last_outcome: Option, + pub running: bool, +} + #[cfg(test)] mod tests { use super::*; @@ -217,7 +275,7 @@ mod tests { fn name(&self) -> &str { &self.name } - async fn run(&self) -> JobOutcome { + async fn run(&self, _args: &JobRunArgs) -> JobOutcome { JobOutcome::ok(0) } } @@ -299,13 +357,20 @@ mod tests { let reg = Arc::new(JobRegistry::new()); reg.register(handler("gc"), None, None).await.unwrap(); - let outcome = reg.trigger("gc").await.expect("job exists"); + let outcome = reg + .trigger("gc", &JobRunArgs::default()) + .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()); + assert!( + reg.trigger("nope", &JobRunArgs::default()) + .await + .is_none() + ); } } diff --git a/src/infrastructure/scheduler/types.rs b/src/infrastructure/scheduler/types.rs index bdc9b438..88c2fecb 100644 --- a/src/infrastructure/scheduler/types.rs +++ b/src/infrastructure/scheduler/types.rs @@ -9,6 +9,27 @@ use std::fmt; use serde::{Deserialize, Serialize}; +/// Per-dispatch parameters passed from the caller (scheduler tick or +/// admin trigger) into [`JobHandler::run`](super::handler::JobHandler::run). +/// +/// Deliberately a struct — not a bare `bool` — so we don't churn every +/// handler signature the next time a job needs another knob. Grows by +/// addition; renaming a field is a breaking change to admin scripts +/// that pass query params, so treat like SQL columns. +/// +/// **Handlers that don't understand a given arg silently ignore it.** +/// No error path just because a caller set an unused flag — that would +/// leak per-job semantics into callers who don't need to know. +/// +/// Semantics of `force`, per job: +/// - `dedup_gc` — skip the orphan grace window (grace = 0). +/// - `grant_cleanup` — grace = 0. +/// - Others (trash_cleanup, storage_reconcile, …) — ignored. +#[derive(Debug, Clone, Default)] +pub struct JobRunArgs { + pub force: bool, +} + /// Uniform outcome the supervisor logs and stores for every job dispatch. /// /// Two variants, deliberately. Distinguishing *why* a job failed diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index 45852e21..ff04cc2d 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -3124,7 +3124,7 @@ impl DedupPort for DedupService { /// Registered name for the dedup GC job. Stable identifier used in /// log lines, `admin.background_runs.job_name` (when Part 2 lands), -/// and admin URLs (`POST /api/admin/internal/trigger-job/dedup_gc`). +/// and admin URLs (`POST /api/admin/jobs/dedup_gc/trigger`). pub const DEDUP_GC_JOB_NAME: &str = "dedup_gc"; #[async_trait::async_trait] @@ -3136,7 +3136,7 @@ impl crate::infrastructure::scheduler::JobHandler for DedupService { /// Runs one `garbage_collect` sweep — the same reclamation that /// `TrashCleanupService` invokes inline as its tail step, exposed /// through the scheduler so operators can trigger it uniformly via - /// `POST /api/admin/internal/trigger-job/dedup_gc`. + /// `POST /api/admin/jobs/dedup_gc/trigger`. /// /// Registered with `interval = None` (on-demand only): the periodic /// tick belongs to trash cleanup, whose sweep already runs GC as @@ -3148,12 +3148,28 @@ impl crate::infrastructure::scheduler::JobHandler for DedupService { /// `count` reports blobs reclaimed; `extra.bytes_reclaimed` reports /// the freed disk. GC returning `(0, 0)` is normal — it means trash /// cleanup already reaped everything. - async fn run(&self) -> crate::infrastructure::scheduler::JobOutcome { + /// + /// `args.force = true` skips the orphan grace window + /// (`garbage_collect_force` — grace_secs = 0), matching the legacy + /// `POST /admin/internal/trigger-gc?force=true` semantics. Unsafe + /// under concurrent uploads: only reachable through the admin + /// endpoint and only intentionally used by tests + operator + /// diagnostic sessions. + async fn run( + &self, + args: &crate::infrastructure::scheduler::JobRunArgs, + ) -> crate::infrastructure::scheduler::JobOutcome { use crate::infrastructure::scheduler::JobOutcome; - match self.garbage_collect().await { - Ok((items, bytes)) => { - JobOutcome::ok_with(items, serde_json::json!({ "bytes_reclaimed": bytes })) - } + let result = if args.force { + self.garbage_collect_force().await + } else { + self.garbage_collect().await + }; + match result { + Ok((items, bytes)) => JobOutcome::ok_with( + items, + serde_json::json!({ "bytes_reclaimed": bytes, "forced": args.force }), + ), Err(e) => JobOutcome::Err(format!("dedup GC failed: {e}")), } } diff --git a/src/infrastructure/services/grant_cleanup_service.rs b/src/infrastructure/services/grant_cleanup_service.rs index 64de01d5..80598685 100644 --- a/src/infrastructure/services/grant_cleanup_service.rs +++ b/src/infrastructure/services/grant_cleanup_service.rs @@ -23,7 +23,7 @@ use tracing::{error, info}; use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::common::errors::DomainError; -use crate::infrastructure::scheduler::{JobHandler, JobOutcome}; +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRunArgs}; use crate::infrastructure::services::pg_acl_engine::PgAclEngine; use async_trait::async_trait; @@ -112,19 +112,26 @@ impl JobHandler for GrantCleanupService { GRANT_CLEANUP_JOB_NAME } - /// Runs one purge with the configured grace window. `count` on the - /// returned `JobOutcome::Ok` is the number of `role_grants` rows - /// physically deleted; `extra.grace_days` records which grace was - /// applied so admin listings can see it without a second lookup. + /// Runs one purge. `count` on the returned `JobOutcome::Ok` is + /// the number of `role_grants` rows physically deleted; + /// `extra.grace_days` records which grace was applied so admin + /// listings can see it without a second lookup. /// - /// Admin `?force=true` (grace = 0) does NOT come through here — - /// that path calls `purge(Some(0))` directly on the shared - /// `Arc` from the handler. - async fn run(&self) -> JobOutcome { - match self.purge(None).await { - Ok(count) => { - JobOutcome::ok_with(count, serde_json::json!({ "grace_days": self.grace_days })) - } + /// `args.force = true` collapses the grace window to zero for + /// this run only — matches the legacy + /// `POST /admin/internal/trigger-grant-cleanup?force=true` shape. + /// The configured `self.grace_days` is not mutated. + async fn run(&self, args: &JobRunArgs) -> JobOutcome { + let grace_override = if args.force { Some(0) } else { None }; + let effective_grace = grace_override.unwrap_or(self.grace_days); + match self.purge(grace_override).await { + Ok(count) => JobOutcome::ok_with( + count, + serde_json::json!({ + "grace_days": effective_grace, + "forced": args.force, + }), + ), Err(e) => JobOutcome::Err(format!("grant cleanup failed: {e}")), } } diff --git a/src/infrastructure/services/trash_cleanup_service.rs b/src/infrastructure/services/trash_cleanup_service.rs index 96d494ab..e008efa5 100644 --- a/src/infrastructure/services/trash_cleanup_service.rs +++ b/src/infrastructure/services/trash_cleanup_service.rs @@ -6,7 +6,7 @@ use tracing::{debug, error, info, instrument}; use crate::common::errors::Result; use crate::domain::repositories::trash_repository::TrashRepository; use crate::infrastructure::repositories::pg::trash_db_repository::TrashDbRepository; -use crate::infrastructure::scheduler::{JobHandler, JobOutcome}; +use crate::infrastructure::scheduler::{JobHandler, JobOutcome, JobRunArgs}; use crate::infrastructure::services::dedup_service::DedupService; use async_trait::async_trait; @@ -165,7 +165,11 @@ impl JobHandler for TrashCleanupService { /// /// Failure of the trash sweep itself → `Err`. GC failure alone is /// non-fatal and stays logged only. - async fn run(&self) -> JobOutcome { + /// + /// `args.force` is ignored — trash cleanup has no acceleration + /// concept (retention windows are per-item metadata, not a runtime + /// knob). + async fn run(&self, _args: &JobRunArgs) -> JobOutcome { match self.run_once().await { Ok(stats) => { let removed = stats.files_purged + stats.folders_purged; diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index d9442e13..d739f578 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -155,6 +155,12 @@ pub fn admin_routes() -> Router> { "/internal/trigger-grant-cleanup", post(internal_trigger_grant_cleanup), ) + // JobRegistry admin surface — production, always-on, + // audit-logged. See `docs/plan/job-registry.md` §Cross-cutting. + // The `/internal/trigger-*` shims above will be retired in a + // follow-up PR (deprecated forwards to these endpoints). + .route("/jobs", get(list_jobs)) + .route("/jobs/{name}/trigger", post(trigger_job)) // Drives — admin-wide view (distinct from `/api/drives` which // is filtered to the caller's role grants). .route("/drives", get(list_all_drives)) @@ -2298,3 +2304,98 @@ pub async fn internal_trigger_grant_cleanup( ) .into_response() } + +// ───────────────────────────────────────────────────── +// JobRegistry admin surface (`/api/admin/jobs/*`) +// ───────────────────────────────────────────────────── + +/// `GET /api/admin/jobs` — enumerate every registered job with its +/// interval, next-run/last-run timestamps, and last outcome. +/// +/// Production endpoint (no `OXICLOUD_ENABLE_ADMIN_INTERNAL_ENDPOINTS` +/// gate). Read-only, so no audit line — the standard admin-middleware +/// auth check is enough. +#[utoipa::path( + get, + path = "/api/admin/jobs", + responses( + (status = 200, description = "Jobs listed"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required"), + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn list_jobs(State(state): State>) -> impl IntoResponse { + let summary = state.core.job_registry.snapshot().await; + (StatusCode::OK, Json(summary)).into_response() +} + +/// Query parameters for `POST /api/admin/jobs/{name}/trigger`. +/// +/// `force=true` requests acceleration semantics from handlers that +/// support it (dedup_gc → grace = 0, grant_cleanup → grace = 0). +/// Silently ignored by handlers that don't (trash_cleanup, +/// storage_reconcile). +#[derive(serde::Deserialize)] +pub struct TriggerJobQuery { + #[serde(default)] + pub force: bool, +} + +/// `POST /api/admin/jobs/{name}/trigger` — dispatch one run off-schedule. +/// +/// Returns the job's `JobOutcome` inline. Idempotent under exclusivity: +/// if the previous run is still in flight, the handler returns +/// `Ok { count: 0, extra: { "skipped": "already_running" } }` rather +/// than spawning a parallel dispatch. +/// +/// Emits an audit line before dispatch — bulk-mutation side effects on +/// operator command belong on the audit stream. +#[utoipa::path( + post, + path = "/api/admin/jobs/{name}/trigger", + params(("name" = String, Path, description = "Registered job name")), + responses( + (status = 200, description = "Dispatched; outcome inline"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required"), + (status = 404, description = "Job not registered"), + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn trigger_job( + State(state): State>, + axum::extract::Path(name): axum::extract::Path, + axum::extract::Query(query): axum::extract::Query, +) -> impl IntoResponse { + use crate::infrastructure::scheduler::JobRunArgs; + // Audit line BEFORE dispatch so an operator triggering something + // that then hangs still leaves a trail. + tracing::info!( + target: "audit", + event = "job.trigger", + job = %name, + force = query.force, + "👮🏻‍♂️ Admin triggered job {} (force={})", + name, + query.force, + ); + let args = JobRunArgs { force: query.force }; + match state.core.job_registry.trigger(&name, &args).await { + Some(outcome) => ( + StatusCode::OK, + Json(serde_json::json!({ "ok": true, "outcome": outcome })), + ) + .into_response(), + None => ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ + "error": "job not registered", + "name": name, + })), + ) + .into_response(), + } +}