diff --git a/frontend/src/lib/api/endpoints/adminJobs.ts b/frontend/src/lib/api/endpoints/adminJobs.ts index 5722462e..6d37c5bf 100644 --- a/frontend/src/lib/api/endpoints/adminJobs.ts +++ b/frontend/src/lib/api/endpoints/adminJobs.ts @@ -8,7 +8,7 @@ */ import { apiFetch, apiJson } from '$lib/api/client'; import { getCsrfHeaders } from '$lib/api/csrf'; -import type { Finding, JobOutcome, JobSummary, RunSummary } from '$lib/api/types'; +import type { Finding, JobOutcome, JobParamValues, JobSummary, RunSummary } from '$lib/api/types'; const JSON_HEADERS = { 'Content-Type': 'application/json' }; @@ -60,21 +60,22 @@ export function listJobs(): Promise { } /** - * `POST /api/admin/jobs/{name}/trigger?force=X&deep=X&repair=X` — - * dispatch a job on-demand. + * `POST /api/admin/jobs/{name}/trigger` — dispatch a job on-demand with + * whichever parameters it declares. * - * - `force` bypasses per-tenant idempotency checks (e.g. `trash_cleanup` - * skipping when nothing is due). - * - `deep` opts into slow variants (currently only `storage_consistency`, - * propagated by `consistency_batch` to every child). - * - `repair` opts into corrective action on the refcount consistency - * tenants (`blobs_consistency`, `manifests_consistency`, and - * `consistency_batch` which fans out to both). Content-safe: only the - * stored counter changes to match the auditor's computed value. Race- - * safe: the corrective UPDATE recomputes the auditor formula in the - * same statement, so a concurrent write can't leave a stale value. - * Default `false` preserves discovery-only behaviour — surface a - * confirm-first flow when calling with `repair: true`. + * **Which parameters are valid is the job's answer, not this + * function's.** Read them from `JobSummary.parameters` (each carries a + * `type`, a `default` and the handler's own description) and pass the + * ones the operator chose. Anything undeclared comes back as a 400 + * naming what the job does accept. + * + * This used to take fixed `force` / `deep` / `storage` / `repair` + * options, which meant callers could pass a flag to a job that ignored + * it and get a silent no-op — the panel offered exactly that on several + * jobs. + * + * Omitted parameters take their declared defaults server-side, so `{}` + * is a plain run. * * Throws on 4xx / 5xx with the backend's error message when present. * A 404 means the job name isn't registered — surface that specifically @@ -82,17 +83,23 @@ export function listJobs(): Promise { */ export async function triggerJob( name: string, - opts: { force?: boolean; deep?: boolean; storage?: string; repair?: boolean } = {} + opts: JobParamValues = {} ): Promise { + // Free-form, because the accepted set is the job's to declare + // (`JobSummary.parameters`) — not this function's to enumerate. The + // backend validates: an undeclared name is a 400 listing what the + // job does accept, rather than being silently ignored the way the + // old fixed `force/deep/storage/repair` options were on jobs that + // read none of them. const params = new URLSearchParams(); - if (opts.force) params.set('force', 'true'); - if (opts.deep) params.set('deep', 'true'); - if (opts.repair) params.set('repair', 'true'); - // `storage` scopes tenants that respect JobRunArgs.storage — - // currently blobs_consistency / backend_consistency (probes the - // named entry instead of the live backend). See - // `docs/plan/storage-multi-entry.md` slice 7. - if (opts.storage) params.set('storage', opts.storage); + for (const [key, value] of Object.entries(opts)) { + // Skip `false` so a URL carries only what was asked for — the + // backend applies each parameter's declared default for the rest, + // and an explicit `force=false` would read identically while + // making the audit line noisier. + if (value === false || value === undefined || value === '') continue; + params.set(key, String(value)); + } const q = params.toString(); const url = `/api/admin/jobs/${encodeURIComponent(name)}/trigger${q ? `?${q}` : ''}`; const res = await apiFetch(url, { diff --git a/frontend/src/lib/api/types.ts b/frontend/src/lib/api/types.ts index b3f44d80..200112b7 100644 --- a/frontend/src/lib/api/types.ts +++ b/frontend/src/lib/api/types.ts @@ -632,6 +632,34 @@ export interface PausedRunBrief { */ export type Mutates = 'never' | 'always' | 'on_repair_only'; +/** Wire type of a declared job parameter — `JobParamType` on the backend. */ +export type JobParamType = 'boolean' | 'string' | 'number'; + +/** + * One run parameter a job accepts, declared by the handler itself + * (`JobHandler::parameters()`). + * + * This is how the panel knows which knobs a job actually reads. It used + * to guess: `deep` came from a hardcoded name allowlist here, so a job + * gaining a deep mode needed a frontend release, and a job losing one + * left a button that silently did nothing. `force` was offered on every + * job whether or not it was read. + * + * `default` is the value the run uses when the parameter is omitted — + * `null` for a string with no default. + */ +export interface JobParam { + name: string; + type: JobParamType; + default: boolean | number | string | null; + /** The job's own wording for THIS parameter, for the control's + * tooltip. Absent when the handler left it blank. */ + description?: string; +} + +/** Values for one trigger, keyed by declared parameter name. */ +export type JobParamValues = Record; + export interface JobSummary { name: string; /** One or two sentences on what the job does, in English, authored @@ -644,6 +672,9 @@ export interface JobSummary { * the text is the confirmation copy. Independent of `mutates` — the * thumbnail import jobs are `always` AND repair-capable. */ repair_description?: string; + /** What this job accepts on a trigger. Absent — not `[]` — when the + * job takes none, so "render no controls" is the natural default. */ + parameters?: JobParam[]; interval_ms?: number; next_run_at?: string; last_run_at?: string; @@ -670,12 +701,16 @@ export interface JobSummary { startup?: StartupTrigger; } -/** Flags a job configured in `OXICLOUD_STARTUP_JOBS` runs with. */ +/** + * Parameters a job configured in `OXICLOUD_STARTUP_JOBS` runs with, + * keyed by declared name. + * + * A map for the same reason `JobSummary.parameters` is one: the four + * fixed fields it replaced meant a job growing a parameter silently + * dropped it from the "at boot" pill. + */ export interface StartupTrigger { - force: boolean; - deep: boolean; - repair: boolean; - storage?: string; + params?: JobParamValues; } /** diff --git a/frontend/src/lib/components/AdminJobsPanel.svelte b/frontend/src/lib/components/AdminJobsPanel.svelte index 6c51e637..7b0a0770 100644 --- a/frontend/src/lib/components/AdminJobsPanel.svelte +++ b/frontend/src/lib/components/AdminJobsPanel.svelte @@ -34,7 +34,7 @@ cancelJob, purgeJobRuns } from '$lib/api/endpoints/adminJobs'; - import type { Finding, JobSummary, RunSummary, RunStatus } from '$lib/api/types'; + import type { Finding, JobParam, JobSummary, RunSummary, RunStatus } from '$lib/api/types'; // ─── State ──────────────────────────────────────────────────────── @@ -634,27 +634,43 @@ return null; } - // Jobs that respect `?deep=true`: - // * `consistency_batch` — propagates deep to every child that - // understands it - // * `backend_consistency` — deep mode re-reads + re-hashes every - // matched blob for silent bit-rot detection (severity - // `data_loss`). Full read of storage; can take hours on big - // installs — the "Run" button on the same row does the - // enumeration merge-join only. This was `blobs_consistency` - // until that tenant became database-only. - function supportsDeep(name: string): boolean { - return name === 'consistency_batch' || name === 'backend_consistency'; + // A declared parameter by name, or undefined. + // + // Everything below asks the JOB what it accepts + // (`JobHandler::parameters()` on the backend) rather than deciding + // here. `supportsDeep` used to be a hardcoded name allowlist — + // `consistency_batch || backend_consistency` — which meant a job + // gaining a deep mode needed a frontend release to become reachable, + // and a job losing one left a menu item that silently did nothing. + function paramOf(job: JobSummary, name: string): JobParam | undefined { + return job.parameters?.find((p) => p.name === name); } - // Whether `?repair=true` does anything for this job — declared by the - // handler itself via `repair_description()`, not by a name allowlist - // here. The allowlist this replaces named only the two ref_count - // tenants and silently omitted every repair-capable job added since, - // so the thumbnail imports could not be run in repair mode from the - // panel at all despite supporting it. + function supportsDeep(job: JobSummary): boolean { + return !!paramOf(job, 'deep'); + } + + // Both signals must agree. `parameters` says the run accepts the + // flag; `repair_description` is the confirmation copy, and a repair + // action with no wording would be a destructive click with a blank + // dialog. A job declaring one without the other is a backend bug — + // render nothing rather than guess. function supportsRepair(job: JobSummary): boolean { - return !!job.repair_description; + return !!paramOf(job, 'repair') && !!job.repair_description; + } + + // Booleans the generic menu renders on its own, beyond the two with + // bespoke entries above. This is what makes a newly-declared flag + // appear with no frontend change. + // + // Booleans only: `storage` and any future string/number parameter + // need a value, and the places that supply one (the storage tab's + // audit / migrate actions) already pass it contextually. A generic + // text box in a run menu would be a worse way to ask. + function extraBooleanParams(job: JobSummary): JobParam[] { + return (job.parameters ?? []).filter( + (p) => p.type === 'boolean' && p.name !== 'deep' && p.name !== 'repair' + ); } // What the repair adds, in the handler's own words. The backend owns @@ -872,11 +888,12 @@ {cadenceLabel(job)} {#if job.startup} + {@const bootRepair = job.startup.params?.repair === true} - {job.startup.repair + {bootRepair ? t('admin.jobs.startup_repair', 'at boot · repair') : t('admin.jobs.startup', 'at boot')} @@ -974,7 +991,8 @@ button — no chevron, no menu, no extra width. Preserves one-click discovery for the common case. --> - {@const hasRunVariants = supportsDeep(job.name) || supportsRepair(job)} + {@const hasRunVariants = + supportsDeep(job) || supportsRepair(job) || extraBooleanParams(job).length > 0} {/if} + + {#each extraBooleanParams(job) as p (p.name)} + + {/each} {/if} {/if} diff --git a/src/common/config.rs b/src/common/config.rs index cc3978fc..c240fba1 100644 --- a/src/common/config.rs +++ b/src/common/config.rs @@ -2,8 +2,6 @@ use std::env; use std::path::PathBuf; use std::time::Duration; -use crate::infrastructure::scheduler::JobRunArgs; - /// Cache configuration #[derive(Debug, Clone)] pub struct CacheConfig { @@ -2337,8 +2335,10 @@ pub struct GrantCleanupConfig { pub struct StartupJob { /// Registered job name — must match `JobHandler::name`. pub name: String, - /// Forwarded verbatim to `JobRegistry::trigger`. - pub args: JobRunArgs, + /// Untyped `key=value` pairs, parsed against the job's declared + /// parameters at dispatch. See [`parse_startup_job`] for why the + /// typing cannot happen here. + pub raw_params: Vec<(String, String)>, } /// Parse one `OXICLOUD_STARTUP_JOBS` entry: `name`, or @@ -2364,39 +2364,24 @@ fn parse_startup_job(raw: &str) -> Result { return Err("empty job name".to_string()); } - let mut job = StartupJob { - name: name.to_string(), - args: JobRunArgs::default(), - }; - + // Raw pairs only. Config is parsed long before the job registry + // exists, so the declaration is not reachable here — typing and + // validation happen at dispatch (`di.rs`), which is also where an + // unknown job NAME is already caught with a boot panic. Both + // failures therefore surface at the same moment and in the same + // shape, rather than one at parse and one at dispatch. + let mut raw_params = Vec::new(); for pair in query.split('&').filter(|p| !p.is_empty()) { let (key, value) = pair .split_once('=') .ok_or_else(|| format!("`{pair}` is not key=value (job `{name}`)"))?; - // Booleans accept only `true`/`false` — the same rule the HTTP - // trigger enforces, so a value that works in one place works in - // the other. See memory `bug_axum_query_bool_only_accepts_true_false`. - let as_bool = || match value { - "true" => Ok(true), - "false" => Ok(false), - other => Err(format!( - "`{key}={other}` on job `{name}`: expected true or false" - )), - }; - match key { - "force" => job.args.force = as_bool()?, - "deep" => job.args.deep = as_bool()?, - "repair" => job.args.repair = as_bool()?, - "storage" => job.args.storage = Some(value.to_string()), - other => { - return Err(format!( - "unknown flag `{other}` on job `{name}`: expected force, deep, repair \ - or storage" - )); - } - } + raw_params.push((key.to_string(), value.to_string())); } - Ok(job) + + Ok(StartupJob { + name: name.to_string(), + raw_params, + }) } /// What runs at boot when `OXICLOUD_STARTUP_JOBS` is unset. @@ -4032,31 +4017,39 @@ mod tests { assert_eq!(rl.delta_upload_window_secs, 60); } + /// Helper: the raw value for `key`, or `None`. + fn raw<'a>(job: &'a StartupJob, key: &str) -> Option<&'a str> { + job.raw_params + .iter() + .find(|(k, _)| k == key) + .map(|(_, v)| v.as_str()) + } + #[test] - fn startup_job_parses_name_and_flags() { + fn startup_job_parses_name_and_params() { let jobs = parse_startup_jobs( "thumb_derived_import?repair=true, thumb_attached_import ,blobs_consistency?deep=true&force=false", ); assert_eq!(jobs.len(), 3); assert_eq!(jobs[0].name, "thumb_derived_import"); - assert!(jobs[0].args.repair); - assert!(!jobs[0].args.deep); + assert_eq!(raw(&jobs[0], "repair"), Some("true")); - // Bare name → all flags default off, which is the discovery-only - // run. Naming a migration job without `repair` imports and stops. + // Bare name → no params at all, so every declared default + // applies. Naming a migration job without `repair` imports and + // stops, which is the discovery-only run. assert_eq!(jobs[1].name, "thumb_attached_import"); - assert!(!jobs[1].args.repair); + assert!(jobs[1].raw_params.is_empty()); - assert!(jobs[2].args.deep); - assert!(!jobs[2].args.force); + assert_eq!(raw(&jobs[2], "deep"), Some("true")); + assert_eq!(raw(&jobs[2], "force"), Some("false")); } #[test] fn startup_job_accepts_storage_scope() { let jobs = parse_startup_jobs("backend_consistency?storage=s3_prod&deep=true"); - assert_eq!(jobs[0].args.storage.as_deref(), Some("s3_prod")); - assert!(jobs[0].args.deep); + assert_eq!(raw(&jobs[0], "storage"), Some("s3_prod")); + assert_eq!(raw(&jobs[0], "deep"), Some("true")); } #[test] @@ -4087,27 +4080,29 @@ mod tests { "transcode_import" ] ); - assert!(jobs.iter().all(|j| j.args.repair)); - assert!(jobs.iter().all(|j| !j.args.deep && !j.args.force)); + assert!(jobs.iter().all(|j| raw(j, "repair") == Some("true"))); + assert!( + jobs.iter() + .all(|j| raw(j, "deep").is_none() && raw(j, "force").is_none()) + ); } - /// A misspelled flag must not parse. Silently ignoring `repare=true` - /// leaves the job in discovery-only mode while the operator believes - /// the tier is draining — a failure that surfaces months later as - /// "the migration never finished", with nothing pointing at the - /// config line. + /// A misspelled or non-boolean parameter must still be fatal at boot + /// — silently ignoring `repare=true` leaves the job in discovery-only + /// mode while the operator believes the tier is draining, a failure + /// that surfaces months later as "the migration never finished" with + /// nothing pointing at the config line. + /// + /// **That check moved rather than went away.** It now runs in + /// `di.rs`, against the job's declared parameters, because only there + /// is the registry built — which also means the error names the + /// job's REAL parameters instead of a hardcoded list. Parsing here + /// deliberately accepts any `key=value`; see + /// `JobRunArgs::from_declared` and its tests for the rejection. #[test] - #[should_panic(expected = "unknown flag `repare`")] - fn startup_job_rejects_a_misspelled_flag() { - parse_startup_jobs("thumb_derived_import?repare=true"); - } - - /// Booleans take only true/false — the same rule the HTTP trigger - /// enforces, so a value that works in one place works in the other. - #[test] - #[should_panic(expected = "expected true or false")] - fn startup_job_rejects_a_non_boolean_flag_value() { - parse_startup_jobs("thumb_derived_import?repair=yes"); + fn startup_job_defers_parameter_validation_to_dispatch() { + let jobs = parse_startup_jobs("thumb_derived_import?repare=true"); + assert_eq!(raw(&jobs[0], "repare"), Some("true")); } #[test] diff --git a/src/common/di.rs b/src/common/di.rs index 0ed2fced..fe88a3fa 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -2922,35 +2922,54 @@ impl AppServiceFactory { if !self.config.startup_jobs.is_empty() { let mut planned = Vec::with_capacity(self.config.startup_jobs.len()); for job in &self.config.startup_jobs { - if app_state.core.job_registry.get(&job.name).await.is_none() { + let Some(declared) = app_state.core.job_registry.parameters_of(&job.name).await + else { panic!( "OXICLOUD_STARTUP_JOBS names `{}`, which is not a registered job. \ Check the spelling against GET /api/admin/jobs.", job.name ); - } - planned.push(job.clone()); + }; + // Same fail-fast rule as the unknown-name panic above, and + // for the same reason: a typo'd `?repare=true` would leave + // a migration importing forever in discovery mode while the + // operator believed the tier was draining. The declaration + // is only reachable here, after the registry is built — + // config parsing kept the pairs untyped. + let args = crate::infrastructure::scheduler::JobRunArgs::from_declared( + declared, + job.raw_params.iter().map(|(k, v)| (k.as_str(), v.as_str())), + ) + .unwrap_or_else(|e| { + panic!("OXICLOUD_STARTUP_JOBS entry `{}`: {e}", job.name); + }); + planned.push((job.name.clone(), args)); } let registry = app_state.core.job_registry.clone(); tokio::spawn(async move { - for job in planned { + for (job_name, args) in planned { // Audited, not merely logged: a startup job may delete // files, and "who asked for this" must be answerable // afterwards. The answer is the configuration, which is // exactly what this line records. + // + // Rendered from the parsed args rather than naming each + // parameter, so a job growing one cannot end up + // dispatched with something the audit trail omits. + let params_desc = args + .iter() + .filter_map(|(k, v)| v.to_param_string().map(|s| format!("{k}={s}"))) + .collect::>() + .join(", "); tracing::info!( target: "audit", event = "job.startup_trigger", - job = %job.name, - force = job.args.force, - deep = job.args.deep, - repair = job.args.repair, - storage = ?job.args.storage, - "👮🏻‍♂️ dispatching `{}` from OXICLOUD_STARTUP_JOBS", - job.name, + job = %job_name, + params = %params_desc, + "👮🏻‍♂️ dispatching `{job_name}` from OXICLOUD_STARTUP_JOBS ({params_desc})", ); - match registry.trigger(&job.name, &job.args).await { + match registry.trigger(&job_name, &args).await { // Debug, not info. The engine already logs every // dispatch as `job.run` with the outcome and timing — // that is the point of routing through `trigger` @@ -2962,10 +2981,9 @@ impl AppServiceFactory { Some(outcome) => tracing::debug!( target: "oxicloud::scheduler", event = "job.startup_completed", - job = %job.name, + job = %job_name, outcome = outcome.kind(), - "startup job `{}` finished ({})", - job.name, + "startup job `{job_name}` finished ({})", outcome.kind(), ), // Unreachable — the name was resolved above, and @@ -2974,10 +2992,9 @@ impl AppServiceFactory { None => tracing::error!( target: "oxicloud::scheduler", event = "job.startup_vanished", - job = %job.name, - "startup job `{}` disappeared from the registry between \ + job = %job_name, + "startup job `{job_name}` disappeared from the registry between \ validation and dispatch", - job.name, ), } } diff --git a/src/infrastructure/scheduler/engine.rs b/src/infrastructure/scheduler/engine.rs index ee122104..58dc0b5c 100644 --- a/src/infrastructure/scheduler/engine.rs +++ b/src/infrastructure/scheduler/engine.rs @@ -165,7 +165,13 @@ pub(super) async fn dispatch(name: &str, entry: Arc, args: &JobRunArgs // 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 args_owned = args.clone(); + // Normalise HERE, the one funnel every dispatch passes through, so a + // handler always sees its declared parameters with their declared + // defaults — whatever the caller built. The periodic tick in + // particular passes an empty `JobRunArgs::default()`, which would + // otherwise read a `default: true` parameter as false on every + // scheduled run. See `JobRunArgs::normalized_for`. + let args_owned = args.normalized_for(handler.parameters()); let join = tokio::spawn(async move { handler.run(&args_owned).await }); let (outcome, cause) = match entry.timeout { diff --git a/src/infrastructure/scheduler/handler.rs b/src/infrastructure/scheduler/handler.rs index 8135db26..51ef3978 100644 --- a/src/infrastructure/scheduler/handler.rs +++ b/src/infrastructure/scheduler/handler.rs @@ -9,7 +9,7 @@ use async_trait::async_trait; -use super::types::{JobOutcome, JobRunArgs, Mutates}; +use super::types::{JobOutcome, JobParam, JobRunArgs, Mutates}; /// Implemented by every service that wants to run on a fixed interval /// through the periodic scheduler. @@ -128,4 +128,19 @@ pub trait JobHandler: Send + Sync { fn repair_description(&self) -> Option<&'static str> { None } + + /// The run parameters this job accepts. + /// + /// Defaults to none, which is correct for most jobs and is now + /// *enforced*: triggering a job with a parameter it does not declare + /// is a 400 naming what it does accept, rather than being silently + /// ignored. A job that reads `args.get_bool("repair")` without + /// declaring `repair` will therefore always see `false` — declare + /// and read together. + /// + /// See [`JobParam`] for why this replaced the fixed + /// force/deep/repair/storage struct. + fn parameters(&self) -> &'static [JobParam] { + &[] + } } diff --git a/src/infrastructure/scheduler/mod.rs b/src/infrastructure/scheduler/mod.rs index 7e16b270..a78b9310 100644 --- a/src/infrastructure/scheduler/mod.rs +++ b/src/infrastructure/scheduler/mod.rs @@ -40,4 +40,7 @@ pub use recoverable::{ pub use registry::{ JobEntry, JobRegistry, JobSummary, PausedRunBrief, RegisterError, StartupTrigger, }; -pub use types::{ErrCause, JobOutcome, JobRunArgs, Mutates}; +pub use types::{ + ErrCause, JobOutcome, JobParam, JobParamDefault, JobParamType, JobParamValue, JobRunArgs, + Mutates, +}; diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index 99df9b55..98db6dea 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -48,7 +48,7 @@ use uuid::Uuid; use crate::common::errors::DomainError; use super::handler::JobHandler; -use super::types::{JobOutcome, JobRunArgs, Mutates}; +use super::types::{JobOutcome, JobParam, JobRunArgs, Mutates}; // ─── Run status ───────────────────────────────────────────────────────────── @@ -207,53 +207,77 @@ impl RunOutcome { /// passed — see the call site in [`run_or_resume`] for why changing mode /// mid-run is refused. /// -/// Every flag is stored as a string, matching the `params` convention the +/// Every value is stored as a string, matching the `params` convention the /// progress fields already use, and each is read back independently: a run -/// paused before this existed simply has no keys, and each missing one -/// falls back to `false` / `None`. That is the safe direction — a resumed +/// paused before its job declared a parameter simply has no key for it, and +/// the declared default applies. That is the safe direction — a resumed /// legacy run under-acts rather than deleting under a flag nobody gave it. +/// +/// **Driven by `declared`, not by a hardcoded list.** The previous version +/// carried `const FLAGS = ["force", "deep", "repair"]` plus a special case +/// for `storage`, so a job growing a parameter had to remember to edit this +/// function — and forgetting meant the parameter was silently dropped on +/// resume, turning a `?repair=true` migration back into a discovery run +/// after a restart. Iterating the declaration makes that unrepresentable. async fn persist_or_restore_args( store: &dyn JobStore, + declared: &[JobParam], args: &JobRunArgs, is_fresh: bool, ) -> Result { - const FLAGS: [&str; 3] = ["force", "deep", "repair"]; - if is_fresh { - for (key, value) in FLAGS.iter().zip([args.force, args.deep, args.repair]) { - let v = if value { "true" } else { "false" }; - store - .set_string_param(key, v) - .await - .map_err(|e| format!("persist `{key}` to params: {e}"))?; + // Filter to what THIS job declares rather than persisting whatever + // the caller handed over. `consistency_batch` forwards its own args + // verbatim to each sub-job, so without this a tenant's `params` + // would grow the coordinator's keys — `deep` on a job that has no + // deep mode — and the run-detail view would claim a mode the job + // never had. + let mut effective = std::collections::BTreeMap::new(); + for p in declared { + let value = args + .iter() + .find(|(k, _)| *k == p.name) + .map(|(_, v)| v.clone()) + .unwrap_or_else(|| p.default.to_value()); + // A `None` string is absent rather than empty, so a run that + // did not scope itself does not grow a key claiming it did. + if let Some(v) = value.to_param_string() { + store + .set_string_param(p.name, &v) + .await + .map_err(|e| format!("persist `{}` to params: {e}", p.name))?; + } + effective.insert(p.name.to_string(), value); } - // `storage` is absent rather than empty when unset, so a run that - // did not scope itself does not grow a key claiming it did. - if let Some(name) = &args.storage { - store - .set_string_param("storage", name) - .await - .map_err(|e| format!("persist `storage` to params: {e}"))?; - } - return Ok(args.clone()); + return Ok(JobRunArgs::new(effective)); } - let mut restored = JobRunArgs::default(); - for (key, slot) in FLAGS.iter().zip([ - &mut restored.force, - &mut restored.deep, - &mut restored.repair, - ]) { - *slot = match store.get_string_param(key).await { - Ok(v) => v.as_deref() == Some("true"), - Err(e) => return Err(format!("read `{key}` from params: {e}")), + let mut restored = std::collections::BTreeMap::new(); + for p in declared { + let stored = store + .get_string_param(p.name) + .await + .map_err(|e| format!("read `{}` from params: {e}", p.name))?; + let value = match stored { + // A value this job wrote itself, so a parse failure means the + // row was hand-edited or the parameter changed type between + // releases. Fall back to the default rather than failing the + // resume — losing the flag is recoverable, refusing to resume a + // half-finished migration is not. + Some(raw) => p.parse_value(&raw).unwrap_or_else(|_| { + tracing::warn!( + target: "oxicloud::scheduler", + param = p.name, + raw = %raw, + "stored job parameter does not parse as its declared type; using the default" + ); + p.default.to_value() + }), + None => p.default.to_value(), }; + restored.insert(p.name.to_string(), value); } - restored.storage = store - .get_string_param("storage") - .await - .map_err(|e| format!("read `storage` from params: {e}"))?; - Ok(restored) + Ok(JobRunArgs::new(restored)) } // ─── Traits — implementor + port ──────────────────────────────────────────── @@ -336,6 +360,19 @@ pub trait RecoverableJobHandler: Send + Sync { None } + /// The run parameters this job accepts. See + /// [`JobHandler::parameters`](super::handler::JobHandler::parameters). + /// + /// Matters more here than for a plain job: `run_or_resume` persists + /// these so a Paused run resumes with the same parameters it started + /// under. The engine iterates this declaration to do it, so an + /// undeclared parameter is not merely ignored — it is lost across a + /// resume, which is how a `?repair=true` migration could come back + /// as discovery-only after a restart. + fn parameters(&self) -> &'static [JobParam] { + &[] + } + /// Long-running scan. See trait-level doc for the contract. /// /// `store` — bound to THIS run (a single row in @@ -877,7 +914,7 @@ pub async fn run_or_resume( // resume would apply it to the remaining entries only, producing a run // that half-deleted — the honest way to change your mind is to cancel // and start fresh. - let args = match persist_or_restore_args(&*store, args, is_fresh).await { + let args = match persist_or_restore_args(&*store, job.parameters(), args, is_fresh).await { Ok(effective) => effective, Err(e) => { // Fail the run rather than guess. Proceeding would mean acting @@ -1139,6 +1176,15 @@ impl JobHandler for RecoverableAdapter { // to `GET /api/admin/jobs`. Silently returning the JobHandler defaults // here would leave every recoverable job undescribed and reported as // read-only — including ones that delete files. + // + // EVERY metadata method the tenant can declare belongs here. Adding + // one to `RecoverableJobHandler` without adding it below compiles + // cleanly — both traits have defaults — and the tenant's value is + // then simply lost. `parameters` shipped that way for exactly one + // boot: the default `&[]` made the trigger endpoint reject + // `?repair=true` on the very jobs that declare it, and + // `OXICLOUD_STARTUP_JOBS` panicked at startup with "this job accepts + // none". Pinned by `adapter_forwards_tenant_metadata`. fn description(&self) -> &'static str { self.inner.description() } @@ -1148,6 +1194,9 @@ impl JobHandler for RecoverableAdapter { fn repair_description(&self) -> Option<&'static str> { self.inner.repair_description() } + fn parameters(&self) -> &'static [JobParam] { + self.inner.parameters() + } } // ─── Ergonomics: JobRegistry extension for recoverable jobs ───────────────── @@ -1689,6 +1738,10 @@ mod tests { fn repair_description(&self) -> Option<&'static str> { Some("fixes the thing") } + fn parameters(&self) -> &'static [JobParam] { + const PARAMS: &[JobParam] = &[JobParam::boolean("repair", false, "fix the thing")]; + PARAMS + } } let provider: Arc = Arc::new(MemProvider::new()); @@ -1698,6 +1751,19 @@ mod tests { assert_eq!(as_handler.description(), "walks a thing"); assert_eq!(as_handler.mutates(), Mutates::OnRepairOnly); assert_eq!(as_handler.repair_description(), Some("fixes the thing")); + + // Regression: this one was NOT forwarded when `parameters` was + // added, and both traits having defaults meant it compiled + // silently. The registry then saw `&[]`, so the trigger endpoint + // rejected `?repair=true` on the jobs that declare it and + // `OXICLOUD_STARTUP_JOBS=thumb_derived_import?repair=true` + // panicked at boot with "this job accepts none". + assert_eq!( + as_handler.parameters().len(), + 1, + "tenant parameters must reach the registry through the adapter" + ); + assert_eq!(as_handler.parameters()[0].name, "repair"); } #[tokio::test] diff --git a/src/infrastructure/scheduler/registry.rs b/src/infrastructure/scheduler/registry.rs index e698261c..951c23c8 100644 --- a/src/infrastructure/scheduler/registry.rs +++ b/src/infrastructure/scheduler/registry.rs @@ -20,7 +20,7 @@ use serde::Serialize; use tokio::sync::{RwLock, Semaphore}; use super::handler::JobHandler; -use super::types::{JobOutcome, JobRunArgs, Mutates}; +use super::types::{JobOutcome, JobParam, JobParamValue, JobRunArgs, Mutates}; /// A registered job plus its runtime state. Held as `Arc` /// inside the registry so the engine can hold a snapshot across an @@ -211,6 +211,18 @@ impl JobRegistry { guard.iter().map(|(k, v)| (k.clone(), v.clone())).collect() } + /// The parameters `name` declares, or `None` when no such job is + /// registered. + /// + /// Callers need this BEFORE dispatch: raw query strings can only be + /// parsed against the declaration, and an undeclared parameter has + /// to be rejected rather than dropped. Returning `None` lets the + /// caller answer 404 for an unknown job without a second lookup. + pub async fn parameters_of(&self, name: &str) -> Option<&'static [JobParam]> { + let guard = self.entries.read().await; + guard.get(name).map(|e| e.handler.parameters()) + } + /// 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 @@ -230,6 +242,7 @@ impl JobRegistry { description: entry.handler.description(), mutates: entry.handler.mutates(), repair_description: entry.handler.repair_description(), + parameters: entry.handler.parameters(), interval_ms: entry.interval.map(|d| d.as_millis() as u64), next_run_at: state.next_run_at, last_run_at, @@ -331,6 +344,11 @@ pub struct JobSummary { /// is the confirmation text. #[serde(skip_serializing_if = "Option::is_none")] pub repair_description: Option<&'static str>, + /// What this job accepts on a trigger. The panel renders exactly + /// these — previously it showed the same fixed checkboxes on every + /// job, most of which the job ignored with no way to tell. + #[serde(skip_serializing_if = "<[_]>::is_empty")] + pub parameters: &'static [JobParam], #[serde(skip_serializing_if = "Option::is_none")] pub interval_ms: Option, #[serde(skip_serializing_if = "Option::is_none")] @@ -360,19 +378,18 @@ pub struct JobSummary { pub startup: Option, } -/// The flags a job configured in `OXICLOUD_STARTUP_JOBS` runs with. +/// The parameters a job configured in `OXICLOUD_STARTUP_JOBS` runs with. /// -/// Mirrors `JobRunArgs` on the wire rather than embedding it, because -/// this is an API shape the admin panel switches on, and `JobRunArgs` -/// is an internal dispatch type free to change without a frontend -/// release. +/// A map keyed by parameter name, for the same reason `JobRunArgs` is: +/// the four named fields it used to carry meant a job growing a +/// parameter silently dropped it from the panel's "at boot" pill. +/// +/// Still a distinct type rather than `JobRunArgs` itself — this is an +/// API shape the admin panel switches on, and the dispatch type should +/// stay free to change without a frontend release. #[derive(Debug, Clone, Serialize)] pub struct StartupTrigger { - pub force: bool, - pub deep: bool, - pub repair: bool, - #[serde(skip_serializing_if = "Option::is_none")] - pub storage: Option, + pub params: std::collections::BTreeMap, } /// Enough info about a paused recoverable run for the admin panel to @@ -498,6 +515,13 @@ mod tests { fn repair_description(&self) -> Option<&'static str> { Some("also deletes the thing") } + fn parameters(&self) -> &'static [JobParam] { + const PARAMS: &[JobParam] = &[ + JobParam::boolean("force", false, "skip the grace window"), + JobParam::string("storage", "entry to scope to"), + ]; + PARAMS + } } let reg = JobRegistry::new(); @@ -507,6 +531,32 @@ mod tests { assert_eq!(row.description, "does a thing"); assert_eq!(row.mutates, Mutates::Always); assert_eq!(row.repair_description, Some("also deletes the thing")); + assert_eq!(row.parameters.len(), 2); + assert_eq!(row.parameters[0].name, "force"); + + // The wire contract the admin panel renders from. Pinned as JSON + // because the panel switches on these exact strings — `type` + // (not `param_type`), snake_case values, and `default` inlined + // rather than tagged. Renaming any of them is a frontend break, + // the same way renaming a `Mutates` variant is. + let json = serde_json::to_value(row).unwrap(); + assert_eq!( + json["parameters"], + serde_json::json!([ + { + "name": "force", + "type": "boolean", + "default": false, + "description": "skip the grace window" + }, + { + "name": "storage", + "type": "string", + "default": null, + "description": "entry to scope to" + } + ]) + ); // Undeclared jobs stay at the safe defaults so the panel can tell // "read-only" from "not yet described" — empty string, not prose. @@ -516,6 +566,16 @@ mod tests { assert_eq!(bare.description, ""); assert_eq!(bare.mutates, Mutates::Never); assert!(bare.repair_description.is_none()); + // Omitted entirely rather than sent as `[]`, so the panel renders + // no parameter controls at all for a job that takes none. + assert!(bare.parameters.is_empty()); + assert!( + serde_json::to_value(bare) + .unwrap() + .get("parameters") + .is_none(), + "an empty declaration must not reach the wire" + ); } #[tokio::test] diff --git a/src/infrastructure/scheduler/types.rs b/src/infrastructure/scheduler/types.rs index 3a20378a..ab70fc25 100644 --- a/src/infrastructure/scheduler/types.rs +++ b/src/infrastructure/scheduler/types.rs @@ -9,61 +9,155 @@ 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). +/// Per-dispatch parameter values, keyed by the names the job declared +/// in [`JobParam`], passed 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. +/// This carried four fixed fields — `force`, `deep`, `storage`, +/// `repair` — plus a doc block enumerating what each meant for each +/// job, ending in "Others — ignored". That list is gone: the semantics +/// now live on each job's own [`JobParam::description`], next to the +/// code that reads them, where they cannot drift out of date. A job +/// that ignores a parameter no longer *has* it. /// -/// **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. +/// A map rather than a struct because the four fixed fields were +/// hardcoded in six places and silently dropped anything new — see +/// [`JobParam`] for the full story. /// -/// Semantics of `force`, per job: -/// - `dedup_gc` — skip the orphan grace window (grace = 0). -/// - `grant_cleanup` — grace = 0. -/// - Others (trash_cleanup, usage_reconcile, …) — ignored. -/// -/// Semantics of `deep`, per job: -/// - `consistency_batch` — propagate to sub-jobs; only `storage_consistency` -/// currently respects it. Wraps the "run all consistency checks -/// including the slow ones" case behind the same job_name lock as -/// the normal batch (Ed's Option B, 2026-07-29). -/// - `storage_consistency` (future) — enables per-blob re-BLAKE3 (bitrot -/// detection) + mime sniff alongside the fast orphan check. -/// - Others — ignored. -/// -/// Semantics of `storage`, per job (added for the multi-entry storage -/// design — see `docs/plan/storage-multi-entry.md`): -/// - `backend_migration` — the NAME of the target storage entry to -/// copy blobs INTO. Required on a Fresh run (handler refuses -/// without it); ignored on a Resumed run (target read from the -/// persisted `params.target_name`). -/// - `blobs_consistency` / `backend_consistency` (slice 7) — the NAME -/// of the entry to probe instead of the currently-active backend. -/// `None` falls through to the live backend (today's behaviour). -/// - Others — ignored. -/// -/// Semantics of `repair` (added 2026-10-17 for the refcount fix): -/// - `blobs_consistency` / `manifests_consistency` — when `true`, -/// after each `refcount_mismatch` / `manifest_refcount_mismatch` -/// finding is recorded, apply the corrective UPDATE that sets the -/// stored counter to the auditor's computed `actual_ref_count`. -/// Content-safe: the row itself is fine, only the counter is -/// wrong. Race-safe: each UPDATE recomputes the auditor formula -/// in the same statement, so a concurrent write can't leave a -/// stale value. Default `false` preserves discovery-only -/// behaviour. Also propagates through `consistency_batch` to -/// both tenants — one `?repair=true` call fixes both counters. -/// - Others — ignored. +/// **The engine seeds this from the job's declared defaults before +/// overlaying caller values**, so a handler reading a parameter it +/// declared always finds it, of the right type. Reading a parameter the +/// job did NOT declare yields the accessor's fallback — which is a bug +/// in the job, and why `parameters()` and the reads should be edited +/// together. #[derive(Debug, Clone, Default)] pub struct JobRunArgs { - pub force: bool, - pub deep: bool, - pub storage: Option, - pub repair: bool, + values: std::collections::BTreeMap, +} + +impl JobRunArgs { + /// Build from already-parsed values. Callers that have raw wire + /// strings should go through [`JobRunArgs::from_declared`] so the + /// declaration does the parsing and validation. + pub fn new(values: std::collections::BTreeMap) -> Self { + Self { values } + } + + /// Seed from `declared` defaults, then overlay `raw` wire values. + /// + /// This is the single place a caller's strings become typed values, + /// shared by the trigger endpoint, the startup-jobs parser and the + /// resume path — so all three accept exactly the same inputs and + /// reject the same ones. + /// + /// An undeclared name is an error, not a silent drop: `?repare=true` + /// on a job that mutates only under `repair` would otherwise run in + /// discovery mode and report "nothing to do", which reads as success. + pub fn from_declared<'a, I>(declared: &[JobParam], raw: I) -> Result + where + I: IntoIterator, + { + let mut values = std::collections::BTreeMap::new(); + for p in declared { + values.insert(p.name.to_string(), p.default.to_value()); + } + for (key, raw_value) in raw { + let Some(p) = declared.iter().find(|p| p.name == key) else { + return Err(if declared.is_empty() { + format!("unknown parameter '{key}': this job accepts none") + } else { + format!( + "unknown parameter '{key}' (accepted: {})", + declared + .iter() + .map(|p| p.name) + .collect::>() + .join(", ") + ) + }); + }; + values.insert(p.name.to_string(), p.parse_value(raw_value)?); + } + Ok(Self { values }) + } + + /// Reshape to exactly `declared`: every declared parameter present, + /// seeded from its default unless this map already carries it, and + /// anything undeclared dropped. + /// + /// **Applied by `dispatch` to every run**, which is what makes + /// "a handler always sees its declared parameters, with the right + /// defaults" true rather than merely usual. Three callers otherwise + /// bypass the typed constructors and would each be a hole: + /// + /// * the periodic tick, which passes [`JobRunArgs::default()`] — an + /// EMPTY map, so a parameter declared with a non-`false` default + /// would silently read as `false` on every scheduled run; + /// * programmatic triggers like [`JobRunArgs::with_string`], which + /// set one parameter and know nothing of the rest; + /// * `consistency_batch`, which forwards its own args to sub-jobs + /// that declare a different set. + /// + /// Dropping rather than rejecting the undeclared is deliberate here: + /// rejection belongs at the edge, where a human typed the name and + /// can be told. By dispatch the value came from another job's + /// declaration, and silently ignoring it is the whole point. + pub fn normalized_for(&self, declared: &[JobParam]) -> Self { + let mut values = std::collections::BTreeMap::new(); + for p in declared { + let value = self + .values + .get(p.name) + .cloned() + .unwrap_or_else(|| p.default.to_value()); + values.insert(p.name.to_string(), value); + } + Self { values } + } + + /// One string parameter — the shape the storage-scoped programmatic + /// triggers use (`backend_migration`, `backend_rotate`), which know + /// their target and bypass the query-string path. + pub fn with_string(name: &str, value: impl Into) -> Self { + let mut values = std::collections::BTreeMap::new(); + values.insert(name.to_string(), JobParamValue::String(Some(value.into()))); + Self { values } + } + + /// A declared boolean, or `false` when absent. + pub fn get_bool(&self, name: &str) -> bool { + match self.values.get(name) { + Some(JobParamValue::Boolean(b)) => *b, + _ => false, + } + } + + /// A declared string, or `None` when absent or empty. + pub fn get_str(&self, name: &str) -> Option<&str> { + match self.values.get(name) { + Some(JobParamValue::String(Some(s))) if !s.is_empty() => Some(s.as_str()), + _ => None, + } + } + + /// A declared number, or `fallback` when absent. + pub fn get_number(&self, name: &str, fallback: i64) -> i64 { + match self.values.get(name) { + Some(JobParamValue::Number(n)) => *n, + _ => fallback, + } + } + + /// Every value, for the engine's persist path. + pub fn iter(&self) -> impl Iterator { + self.values.iter().map(|(k, v)| (k.as_str(), v)) + } + + /// True when nothing was supplied — used to keep log lines quiet + /// for the common no-parameter dispatch. + pub fn is_empty(&self) -> bool { + self.values.is_empty() + } } /// Uniform outcome the supervisor logs and stores for every job dispatch. @@ -194,10 +288,307 @@ pub enum Mutates { OnRepairOnly, } +/// The type of a declared job parameter, and the shape its value takes +/// on the wire. +/// +/// Three types because that is what the query string and the admin +/// panel can express between them: a checkbox, a text/select input, a +/// number input. Anything richer belongs in the job's own config, not +/// in a per-run parameter. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum JobParamType { + Boolean, + String, + Number, +} + +/// A parameter's declared default. +/// +/// Separate from [`JobParamValue`] so [`JobParam`] contains no `String` +/// and stays const-constructible: a `&'static [JobParam]` literal in a +/// `parameters()` body needs const promotion, which a type with a +/// destructor blocks. +/// +/// No string variant, deliberately — see [`JobParam::string`]: a string +/// parameter that wants a default is usually config in disguise, and +/// `Absent` is what "use the active backend" looks like for `storage`. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[serde(untagged)] +pub enum JobParamDefault { + Boolean(bool), + Number(i64), + /// No default — the parameter is simply absent unless supplied. + Absent, +} + +impl JobParamDefault { + /// The runtime value this default seeds a run with. + pub fn to_value(self) -> JobParamValue { + match self { + Self::Boolean(b) => JobParamValue::Boolean(b), + Self::Number(n) => JobParamValue::Number(n), + Self::Absent => JobParamValue::String(None), + } + } +} + +/// A value for a declared parameter, as supplied for one run. +/// +/// `String` is `Option` because an absent string and an empty one are +/// different for `storage` — absent means "use the active backend", +/// empty would be a nameless entry. +#[derive(Debug, Clone, PartialEq, Serialize)] +#[serde(untagged)] +pub enum JobParamValue { + Boolean(bool), + String(Option), + Number(i64), +} + +impl JobParamValue { + /// The parameter type this value inhabits — used to reject a + /// caller who sends `?deep=7` for a boolean. + pub fn param_type(&self) -> JobParamType { + match self { + Self::Boolean(_) => JobParamType::Boolean, + Self::String(_) => JobParamType::String, + Self::Number(_) => JobParamType::Number, + } + } + + /// Render for the `params` JSONB column, which is `TEXT`-valued so + /// a resumed run can restore whatever the fresh run was given. + pub fn to_param_string(&self) -> Option { + match self { + Self::Boolean(b) => Some(b.to_string()), + Self::Number(n) => Some(n.to_string()), + Self::String(s) => s.clone(), + } + } +} + +/// One parameter a job accepts on a run. +/// +/// # Why jobs declare these +/// +/// The four parameters `force` / `deep` / `repair` / `storage` used to +/// be a fixed struct, and six places hardcoded that same list: the +/// engine's persist/restore, the trigger endpoint's query type, the +/// `OXICLOUD_STARTUP_JOBS` parser, the frontend API wrapper, and the +/// admin panel's checkboxes. Adding a parameter meant editing all of +/// them, and forgetting one meant the parameter was silently dropped — +/// most damagingly by the persist/restore path, where a resumed run +/// would quietly lose it. +/// +/// Worse for operators: the panel showed the same knobs on every job. +/// Only two jobs read `deep` and six read `repair`, so most of those +/// checkboxes did nothing, with no way to tell which. +/// +/// Now each job declares what it accepts. The engine iterates the +/// declaration, the trigger endpoint rejects anything undeclared, and +/// the panel renders exactly the knobs that job reads. +/// +/// # Wire names are a compatibility surface +/// +/// `name` is what `params` rows are keyed by and what the panel +/// switches on, so renaming one breaks existing run history the same +/// way renaming a [`Mutates`] variant would. Add a new parameter +/// rather than repurposing an old one. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +pub struct JobParam { + pub name: &'static str, + #[serde(rename = "type")] + pub param_type: JobParamType, + /// Applied when the caller omits the parameter. The engine seeds + /// every run's args from these before overlaying caller values, so + /// a handler reading a declared parameter always finds it. + pub default: JobParamDefault, + /// One line for the panel's input label. Empty renders bare. + #[serde(skip_serializing_if = "str::is_empty")] + pub description: &'static str, +} + +impl JobParam { + /// A boolean parameter, e.g. `?deep=true`. + pub const fn boolean(name: &'static str, default: bool, description: &'static str) -> Self { + Self { + name, + param_type: JobParamType::Boolean, + default: JobParamDefault::Boolean(default), + description, + } + } + + /// A string parameter with no default, e.g. `?storage=azurite`. + /// + /// No `default` argument: a string parameter that wants one is + /// almost always a config value in disguise. `storage` — the only + /// string parameter today — means "the active backend" when absent, + /// which is a job-side decision, not a default the engine can seed. + pub const fn string(name: &'static str, description: &'static str) -> Self { + Self { + name, + param_type: JobParamType::String, + default: JobParamDefault::Absent, + description, + } + } + + /// A numeric parameter, e.g. `?batch_size=500`. + pub const fn number(name: &'static str, default: i64, description: &'static str) -> Self { + Self { + name, + param_type: JobParamType::Number, + default: JobParamDefault::Number(default), + description, + } + } + + /// Parse a wire value (query string / `OXICLOUD_STARTUP_JOBS` / + /// restored `params` row) according to this parameter's type. + /// + /// Returns `Err` with an operator-facing reason rather than + /// defaulting, so `?deep=yes` fails loudly instead of running a + /// shallow scan the caller did not ask for. + pub fn parse_value(&self, raw: &str) -> Result { + match self.param_type { + JobParamType::Boolean => match raw { + // Deliberately strict — same rule as axum's `Query` + // bool. "yes"/"1"/"on" are the shapes an operator + // reaches for, and silently accepting them here while + // the query layer rejects them would be worse than + // rejecting both. + "true" => Ok(JobParamValue::Boolean(true)), + "false" => Ok(JobParamValue::Boolean(false)), + other => Err(format!( + "'{other}' is not a boolean for parameter '{}' (use true or false)", + self.name + )), + }, + JobParamType::String => Ok(JobParamValue::String(Some(raw.to_string()))), + JobParamType::Number => raw + .parse::() + .map(JobParamValue::Number) + .map_err(|_| format!("'{raw}' is not a number for parameter '{}'", self.name)), + } + } +} + #[cfg(test)] mod tests { use super::*; + const DECLARED: &[JobParam] = &[ + JobParam::boolean("repair", false, "d"), + JobParam::boolean("deep", true, "d"), + JobParam::string("storage", "d"), + JobParam::number("batch", 500, "d"), + ]; + + #[test] + fn declared_defaults_seed_the_run() { + let args = JobRunArgs::from_declared(DECLARED, []).unwrap(); + assert!(!args.get_bool("repair")); + // Not merely "absent reads as false" — a declared `true` default + // must survive, which is the whole reason defaults live in the + // declaration rather than at each read site. + assert!(args.get_bool("deep")); + assert_eq!(args.get_str("storage"), None); + assert_eq!(args.get_number("batch", 0), 500); + } + + #[test] + fn caller_values_overlay_defaults() { + let args = + JobRunArgs::from_declared(DECLARED, [("repair", "true"), ("deep", "false")]).unwrap(); + assert!(args.get_bool("repair")); + assert!(!args.get_bool("deep")); + } + + /// The failure the whole declaration exists to prevent: a typo that + /// silently leaves a destructive job in discovery mode. + #[test] + fn an_undeclared_parameter_is_rejected_and_names_the_real_ones() { + let err = JobRunArgs::from_declared(DECLARED, [("repare", "true")]).unwrap_err(); + assert!(err.contains("repare"), "{err}"); + assert!(err.contains("repair"), "must name what IS accepted: {err}"); + } + + #[test] + fn a_job_declaring_nothing_says_so() { + let err = JobRunArgs::from_declared(&[], [("force", "true")]).unwrap_err(); + assert!(err.contains("accepts none"), "{err}"); + } + + /// Same strictness as the HTTP layer's bool parsing, so a value that + /// works in `OXICLOUD_STARTUP_JOBS` works in the trigger URL. + #[test] + fn booleans_take_only_true_or_false() { + let err = JobRunArgs::from_declared(DECLARED, [("repair", "yes")]).unwrap_err(); + assert!(err.contains("not a boolean"), "{err}"); + assert!(JobRunArgs::from_declared(DECLARED, [("repair", "false")]).is_ok()); + } + + #[test] + fn numbers_must_parse() { + assert!(JobRunArgs::from_declared(DECLARED, [("batch", "x")]).is_err()); + let args = JobRunArgs::from_declared(DECLARED, [("batch", "12")]).unwrap(); + assert_eq!(args.get_number("batch", 0), 12); + } + + /// An empty string is not a storage entry. `get_str` folding it to + /// `None` is what keeps `?storage=` from resolving to a nameless + /// backend rather than the active one. + #[test] + fn an_empty_string_reads_as_absent() { + let args = JobRunArgs::from_declared(DECLARED, [("storage", "")]).unwrap(); + assert_eq!(args.get_str("storage"), None); + } + + /// Reading a parameter the job never declared is a bug in the job, + /// and it fails closed rather than panicking — the accessor's + /// fallback stands in. + #[test] + fn reading_an_undeclared_parameter_falls_back() { + let args = JobRunArgs::from_declared(DECLARED, []).unwrap(); + assert!(!args.get_bool("nonexistent")); + assert_eq!(args.get_number("nonexistent", 7), 7); + } + + /// `dispatch` applies this to every run, so the periodic tick — which + /// passes an EMPTY `JobRunArgs::default()` — still gets the declared + /// defaults. Without it a `default: true` parameter would read as + /// false on every scheduled run and only be right when an operator + /// triggered by hand. + #[test] + fn normalizing_an_empty_args_applies_declared_defaults() { + let args = JobRunArgs::default().normalized_for(DECLARED); + assert!(args.get_bool("deep"), "declared default true must survive"); + assert!(!args.get_bool("repair")); + assert_eq!(args.get_number("batch", 0), 500); + } + + #[test] + fn normalizing_keeps_supplied_values_and_drops_undeclared() { + // As `consistency_batch` forwards: its own `force` reaching a + // sub-job that declares no such thing. + let forwarded = JobRunArgs::new( + [ + ("repair".to_string(), JobParamValue::Boolean(true)), + ("force".to_string(), JobParamValue::Boolean(true)), + ] + .into_iter() + .collect(), + ); + let args = forwarded.normalized_for(DECLARED); + assert!(args.get_bool("repair"), "supplied value survives"); + assert!( + !args.iter().any(|(k, _)| k == "force"), + "an undeclared parameter must not reach the handler or its params row" + ); + } + #[test] fn mutates_serialises_snake_case() { // The admin UI switches on these strings — a rename is a breaking diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 4ba04798..38204047 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -199,6 +199,25 @@ impl RecoverableJobHandler for BackendConsistencyCheck { deleted." } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[ + JobParam::boolean( + "deep", + false, + "Read every matched blob back and re-hash it, catching \ + silent bit-rot. A full read of storage — can take hours.", + ), + JobParam::string( + "storage", + "Name of the storage entry to audit. Absent audits the \ + active backend; naming an entry is how either side of a \ + migration gets audited directly.", + ), + ]; + PARAMS + } + /// Approximate total: on a healthy install every backend blob /// has a `storage.blobs` row, so the DB count is a proxy for /// the backend count. The fraction deviating from 1.0 at run @@ -240,26 +259,40 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // `blobs_consistency` uses — Fresh + args.storage=Some stamps // probed_storage into params; Resumed reads it back so a // mid-audit restart re-uses the same target. - let is_fresh = resume_cursor.is_none(); - let probed_storage: Option = if is_fresh { - let name = args.storage.clone(); - if let Some(n) = &name - && let Err(e) = store.set_string_param(PROBED_STORAGE_PARAM, n).await - { - return RunOutcome::Failed { - message: format!("persist {PROBED_STORAGE_PARAM} to params: {e}"), - }; - } - name - } else { - match store.get_string_param(PROBED_STORAGE_PARAM).await { - Ok(v) => v, - Err(e) => { - return RunOutcome::Failed { - message: format!("read {PROBED_STORAGE_PARAM} from params: {e}"), - }; + // `run_or_resume` persists and restores `storage` for us now, so + // the normal path is a plain read. + // + // The fallback is a MIGRATION concern, not defensiveness. This job + // used to persist the same value under its own + // `probed_storage` key; a run paused before this change has that + // key and no `storage` one. Without the fallback such a run would + // resume against the ACTIVE backend instead of the entry it was + // auditing — silently auditing the wrong thing, which is worse + // than failing. Removable once no pre-upgrade paused runs remain. + let probed_storage: Option = match args.get_str("storage") { + Some(name) => Some(name.to_string()), + None if resume_cursor.is_some() => { + match store.get_string_param(PROBED_STORAGE_PARAM).await { + Ok(legacy) => { + if legacy.is_some() { + tracing::info!( + target: "oxicloud::consistency", + event = "backend_consistency.legacy_storage_param", + run_id = %store.run_id(), + "resumed a run that recorded its target under the pre-declaration \ + `probed_storage` key" + ); + } + legacy + } + Err(e) => { + return RunOutcome::Failed { + message: format!("read {PROBED_STORAGE_PARAM} from params: {e}"), + }; + } } } + None => None, }; let backend: Arc = match &probed_storage { None => self.backend.clone(), @@ -313,30 +346,12 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // that tenant to carry a backend for one flag, which is the // overlap this split removes. // - // Persisted to `params.deep` on a Fresh run so a Resume picks up - // the same mode (a Paused deep scan must not silently continue - // shallow) and the admin run-detail view can show what the scan - // actually verified. Written BEFORE the walk so a crash mid-batch - // still leaves the marker. - let deep = if is_fresh { - let v = if args.deep { "true" } else { "false" }; - if let Err(e) = store.set_string_param("deep", v).await { - return RunOutcome::Failed { - message: format!("failed to persist deep flag to params: {e}"), - }; - } - args.deep - } else { - match store.get_string_param("deep").await { - Ok(Some(v)) => v == "true", - Ok(None) => false, - Err(e) => { - return RunOutcome::Failed { - message: format!("read `deep` from params: {e}"), - }; - } - } - }; + // Persisted to `params.deep` and restored on resume by + // `run_or_resume`, so a Paused deep scan does not silently + // continue shallow and the run-detail view can show what the scan + // actually verified. + let deep = args.get_bool("deep"); + if deep { tracing::info!( target: "oxicloud::consistency", diff --git a/src/infrastructure/services/backend_migration_service.rs b/src/infrastructure/services/backend_migration_service.rs index 9eb56347..aa6b4c15 100644 --- a/src/infrastructure/services/backend_migration_service.rs +++ b/src/infrastructure/services/backend_migration_service.rs @@ -196,6 +196,21 @@ impl RecoverableJobHandler for BackendMigrationService { restarting." } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + // Required in practice, though the declaration cannot express + // that: a Fresh run without it fails with a message naming the + // proper entrypoint, while a Resume legitimately omits it and + // reads the target back from `params.target_name`. + const PARAMS: &[JobParam] = &[JobParam::string( + "storage", + "Name of the storage entry to copy blobs INTO. Required on a \ + fresh run; ignored on a resume, which reuses the recorded \ + target.", + )]; + PARAMS + } + /// Writes bytes to the target backend. Source bytes are left in place — /// the copy is additive, so an aborted migration loses nothing. fn mutates(&self) -> Mutates { @@ -245,7 +260,7 @@ impl RecoverableJobHandler for BackendMigrationService { // into the wrong entry. let is_fresh = resume_cursor.is_none(); let target_name = if is_fresh { - let Some(name) = args.storage.clone() else { + let Some(name) = args.get_str("storage").map(str::to_string) else { return RunOutcome::Failed { message: "backend_migration requires `target_name` on a fresh run — trigger via \ diff --git a/src/infrastructure/services/backend_rotate_service.rs b/src/infrastructure/services/backend_rotate_service.rs index 074a35bf..56a5d9ce 100644 --- a/src/infrastructure/services/backend_rotate_service.rs +++ b/src/infrastructure/services/backend_rotate_service.rs @@ -143,6 +143,17 @@ impl RecoverableJobHandler for BackendRotateService { so re-running after a key change is cheap." } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[JobParam::string( + "storage", + "Name of the storage entry whose blobs to rewrite. Required on \ + a fresh run; ignored on a resume, which reuses the recorded \ + target.", + )]; + PARAMS + } + /// Rewrites blobs **in place**. Unlike a migration this has no additive /// fallback — the previous ciphertext is gone once a blob is rewritten. fn mutates(&self) -> Mutates { @@ -178,7 +189,7 @@ impl RecoverableJobHandler for BackendRotateService { // Resolve target entry name — same shape as `backend_migration`. let is_fresh = resume_cursor.is_none(); let target_name = if is_fresh { - let Some(name) = args.storage.clone() else { + let Some(name) = args.get_str("storage").map(str::to_string) else { return RunOutcome::Failed { message: "backend_rotate requires `target_name` on a fresh run — trigger via \ POST /api/admin/storage/entries/{name}/rotate" diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index 6fdd248c..57b7214c 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -267,6 +267,21 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { ) } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + // No `deep` — re-reading bytes is backend work and moved to + // `backend_consistency`. Declaring it here would put a knob in + // the panel that this job ignores, which is the thing the + // declaration exists to stop. + const PARAMS: &[JobParam] = &[JobParam::boolean( + "repair", + false, + "Rewrite drifted ref_count values to the recomputed truth. \ + Without this the run only reports them.", + )]; + PARAMS + } + /// Definitive count. `storage.blobs` PK scan is index-only; /// even at millions of rows it's sub-second on modern PG. async fn count_total(&self) -> Option { @@ -299,12 +314,6 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { // `backend_consistency`, which finds it in one enumeration pass // instead of one probe per row. // - // Snapshot "is this a Fresh run?" BEFORE the resume_cursor - // match consumes it — otherwise the `is_none()` check later - // borrows a partially-moved value. Fresh = no cursor bytes - // at all; Resumed = cursor bytes present (possibly empty). - let is_fresh = resume_cursor.is_none(); - // Cursor = the last-visited `hash` string, UTF-8-encoded. On // resume, we walk `WHERE hash > $cursor` in ASC order. First // batch: NULL cursor → start from the smallest hash. @@ -340,30 +349,15 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { // matched key pairs worth verifying. A deep flag on this tenant // would be a flag with nothing to do. - // Repair mode persisted to `params.repair` so the admin run-detail - // view can display it. Fresh persists what the trigger asked for; - // Resume reads back so a paused repair scan stays a repair - // scan (a mid-scan crash mustn't silently downgrade to - // discovery-only for the remaining rows). - let repair = if is_fresh { - let v = if args.repair { "true" } else { "false" }; - if let Err(e) = store.set_string_param("repair", v).await { - return RunOutcome::Failed { - message: format!("failed to persist repair flag to params: {e}"), - }; - } - args.repair - } else { - match store.get_string_param("repair").await { - Ok(Some(v)) => v == "true", - Ok(None) => false, - Err(e) => { - return RunOutcome::Failed { - message: format!("read `repair` from params: {e}"), - }; - } - } - }; + // Repair mode is persisted to `params.repair` so the admin + // run-detail view can display it, and restored on resume so a + // paused repair scan stays a repair scan — a mid-scan crash must + // not silently downgrade the remaining rows to discovery-only. + // + // Both happen in `run_or_resume`, for every declared parameter, + // under this same key. This job used to do it itself; that + // duplication is what the parameter declaration removes. + let repair = args.get_bool("repair"); if repair { tracing::info!( diff --git a/src/infrastructure/services/consistency_batch_service.rs b/src/infrastructure/services/consistency_batch_service.rs index d4287375..e1178b0a 100644 --- a/src/infrastructure/services/consistency_batch_service.rs +++ b/src/infrastructure/services/consistency_batch_service.rs @@ -97,6 +97,32 @@ impl JobHandler for ConsistencyBatch { are forwarded to each sub-job." } + /// The union of what its sub-jobs accept, because it forwards + /// verbatim. A sub-job that does not declare one of these simply + /// never sees it — `run_or_resume` filters each dispatch down to that + /// job's own declaration, so forwarding `deep` to a tenant with no + /// deep mode is inert rather than misrecorded. + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[ + JobParam::boolean( + "deep", + false, + "Forwarded to sub-jobs that have a deep mode — currently \ + backend_consistency, which re-reads and re-hashes every \ + blob. Can take hours.", + ), + JobParam::boolean("force", false, "Forwarded to sub-jobs that accept it."), + JobParam::boolean( + "repair", + false, + "Forwarded to every sub-job that can repair, so one call \ + fixes both refcount tenants.", + ), + ]; + PARAMS + } + /// Read-only on a plain run because every tenant it dispatches is, but /// `?repair=true` reaches whichever of them act on it — so the batch /// inherits the strongest mode any sub-job can be put into. @@ -198,9 +224,9 @@ impl JobHandler for ConsistencyBatch { targets.len() as u64, json!({ "per_check": per_check, - "deep": args.deep, - "force": args.force, - "repair": args.repair, + "deep": args.get_bool("deep"), + "force": args.get_bool("force"), + "repair": args.get_bool("repair"), "ok": ok_count, "err": err_count, }), diff --git a/src/infrastructure/services/dedup_service.rs b/src/infrastructure/services/dedup_service.rs index f5faab14..cab37af0 100644 --- a/src/infrastructure/services/dedup_service.rs +++ b/src/infrastructure/services/dedup_service.rs @@ -3873,18 +3873,33 @@ impl crate::infrastructure::scheduler::JobHandler for DedupService { /// the freed disk. GC returning `(0, 0)` is normal — it means trash /// cleanup already reaped everything. /// - /// `args.force = true` skips the orphan grace window + /// `force = true` skips the orphan grace window /// (`garbage_collect_force` — grace_secs = 0). Same semantic as /// `POST /api/admin/jobs/dedup_gc/trigger?force=true`. Unsafe /// under concurrent uploads: only reachable through the admin /// endpoint and only intentionally used by tests + operator /// diagnostic sessions. + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + // A named `const` rather than a bare `&[…]` literal: implicit + // const promotion does not cover `const fn` calls, so the + // literal would be a temporary. Same shape in every job. + const PARAMS: &[JobParam] = &[JobParam::boolean( + "force", + false, + "Skip the orphan grace window. Unsafe under concurrent \ + uploads — it reopens the TOCTOU window the grace closes.", + )]; + PARAMS + } + async fn run( &self, args: &crate::infrastructure::scheduler::JobRunArgs, ) -> crate::infrastructure::scheduler::JobOutcome { use crate::infrastructure::scheduler::JobOutcome; - let result = if args.force { + let force = args.get_bool("force"); + let result = if force { self.garbage_collect_force().await } else { self.garbage_collect().await @@ -3892,7 +3907,7 @@ impl crate::infrastructure::scheduler::JobHandler for DedupService { match result { Ok((items, bytes)) => JobOutcome::ok_with( items, - serde_json::json!({ "bytes_reclaimed": bytes, "forced": args.force }), + serde_json::json!({ "bytes_reclaimed": bytes, "forced": 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 83925a95..8722d168 100644 --- a/src/infrastructure/services/grant_cleanup_service.rs +++ b/src/infrastructure/services/grant_cleanup_service.rs @@ -139,19 +139,31 @@ impl JobHandler for GrantCleanupService { /// `extra.grace_days` records which grace was applied so admin /// listings can see it without a second lookup. /// - /// `args.force = true` collapses the grace window to zero for - /// this run only — same semantic as + /// `force = true` collapses the grace window to zero for this run + /// only — same semantic as /// `POST /api/admin/jobs/grant_cleanup/trigger?force=true`. The /// configured `self.grace_days` is not mutated. + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[JobParam::boolean( + "force", + false, + "Collapse the expiry grace window to zero for this run. \ + The configured grace is not changed.", + )]; + PARAMS + } + async fn run(&self, args: &JobRunArgs) -> JobOutcome { - let grace_override = if args.force { Some(0) } else { None }; + let force = args.get_bool("force"); + let grace_override = if 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, + "forced": force, }), ), Err(e) => JobOutcome::err(format!("grant cleanup failed: {e}")), diff --git a/src/infrastructure/services/manifests_consistency_service.rs b/src/infrastructure/services/manifests_consistency_service.rs index ddd9fe08..0f3aad19 100644 --- a/src/infrastructure/services/manifests_consistency_service.rs +++ b/src/infrastructure/services/manifests_consistency_service.rs @@ -202,6 +202,17 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { ) } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[JobParam::boolean( + "repair", + false, + "Rewrite drifted manifest ref_count values to the recomputed \ + truth. Without this the run only reports them.", + )]; + PARAMS + } + async fn count_total(&self) -> Option { let row: Result<(i64,), sqlx::Error> = sqlx::query_as("SELECT COUNT(*) FROM storage.chunk_manifests") @@ -227,8 +238,6 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { args: &JobRunArgs, resume_cursor: Option>, ) -> RunOutcome { - let is_fresh = resume_cursor.is_none(); - // Cursor: the last `file_hash` as UTF-8. Same convention as // `blobs_consistency`, which also pages a hash-keyed table. let mut cursor: Option = match resume_cursor { @@ -244,33 +253,14 @@ impl RecoverableJobHandler for ManifestsConsistencyCheck { }, }; - // Persist the repair flag into `params.repair` so the admin - // run-detail view can display whether the run was a discovery - // scan or an active repair. Fresh takes it from args; Resume - // reads back so a paused repair scan stays a repair scan (a - // mid-scan crash mustn't silently downgrade the remaining - // rows to discovery-only). Same shape as - // `blobs_consistency_service.rs`'s `deep` handling — see the - // reasoning documented there. - let repair = if is_fresh { - let v = if args.repair { "true" } else { "false" }; - if let Err(e) = store.set_string_param("repair", v).await { - return RunOutcome::Failed { - message: format!("failed to persist repair flag to params: {e}"), - }; - } - args.repair - } else { - match store.get_string_param("repair").await { - Ok(Some(v)) => v == "true", - Ok(None) => false, - Err(e) => { - return RunOutcome::Failed { - message: format!("read `repair` from params: {e}"), - }; - } - } - }; + // Persisted into `params.repair` so the admin run-detail view can + // show whether this was a discovery scan or an active repair, and + // restored on resume so a paused repair scan stays one — a + // mid-scan crash must not silently downgrade the remaining rows. + // + // `run_or_resume` does both, for every declared parameter, under + // this same key. This job used to hand-roll it. + let repair = args.get_bool("repair"); if repair { tracing::info!( diff --git a/src/infrastructure/services/thumb_attached_import_service.rs b/src/infrastructure/services/thumb_attached_import_service.rs index 92917d55..a03586ea 100644 --- a/src/infrastructure/services/thumb_attached_import_service.rs +++ b/src/infrastructure/services/thumb_attached_import_service.rs @@ -185,6 +185,17 @@ impl RecoverableJobHandler for ThumbAttachedImport { ) } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[JobParam::boolean( + "repair", + false, + "Delete each sidecar after its replacement has been read back. \ + Without this the job imports and leaves the originals in place.", + )]; + PARAMS + } + async fn count_total(&self) -> Option { let mut total = 0u64; for size in ThumbnailSize::all() { @@ -229,7 +240,7 @@ impl RecoverableJobHandler for ThumbAttachedImport { // PDF preview has no server-side render path — so it is not // belt-and-braces, it is the only thing between a migration and // permanent loss. - let delete_imported = args.repair; + let delete_imported = args.get_bool("repair"); let mut failed = 0u64; let mut since_checkpoint = 0usize; diff --git a/src/infrastructure/services/thumb_derived_import_service.rs b/src/infrastructure/services/thumb_derived_import_service.rs index 37e67200..30cd773e 100644 --- a/src/infrastructure/services/thumb_derived_import_service.rs +++ b/src/infrastructure/services/thumb_derived_import_service.rs @@ -424,6 +424,18 @@ impl RecoverableJobHandler for ThumbDerivedImport { ) } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[JobParam::boolean( + "repair", + false, + "Delete each sidecar after its replacement has been read back, \ + and remove the directory once empty. Without this the job \ + imports and leaves the originals in place.", + )]; + PARAMS + } + async fn count_total(&self) -> Option { let mut total = 0u64; for size in ThumbnailSize::all() { @@ -449,7 +461,7 @@ impl RecoverableJobHandler for ThumbDerivedImport { // makes the migration self-draining: sidecars are LOCAL disk, so no // release can know whether every instance has finished, whereas each // instance draining itself needs no coordination at all. - let delete_imported = args.repair; + let delete_imported = args.get_bool("repair"); // Cursor is `{size_dir}/{filename}` — the last file completed. Sizes // are walked in `ThumbnailSize::all()` order, and names are sorted // within each, so the pair totally orders the walk. diff --git a/src/infrastructure/services/transcode_import_service.rs b/src/infrastructure/services/transcode_import_service.rs index e1464c54..85c76a1b 100644 --- a/src/infrastructure/services/transcode_import_service.rs +++ b/src/infrastructure/services/transcode_import_service.rs @@ -214,6 +214,18 @@ impl RecoverableJobHandler for TranscodeImport { ) } + fn parameters(&self) -> &'static [crate::infrastructure::scheduler::JobParam] { + use crate::infrastructure::scheduler::JobParam; + const PARAMS: &[JobParam] = &[JobParam::boolean( + "repair", + false, + "Delete each cached transcode after its replacement has been \ + read back and compared byte for byte. Without this the job \ + imports and leaves the originals in place.", + )]; + PARAMS + } + async fn count_total(&self) -> Option { Some(Self::entry_names(&self.variant_dir()).await.len() as u64) } @@ -240,7 +252,7 @@ impl RecoverableJobHandler for TranscodeImport { }, }; - let delete_imported = args.repair; + let delete_imported = args.get_bool("repair"); let dir = self.variant_dir(); let mut imported = 0u64; diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index 455f672f..e6c699ab 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -621,9 +621,9 @@ async fn trigger_backend_migration( ); let registry = state.core.job_registry.clone(); - let args = JobRunArgs { - storage: target_name, - ..JobRunArgs::default() + let args = match target_name { + Some(n) => JobRunArgs::with_string("storage", n), + None => JobRunArgs::default(), }; tokio::spawn(async move { registry.trigger(BACKEND_MIGRATION_JOB_NAME, &args).await; @@ -754,10 +754,7 @@ pub async fn trigger_backend_rotate( ); let registry = state.core.job_registry.clone(); - let args = JobRunArgs { - storage: Some(name.clone()), - ..JobRunArgs::default() - }; + let args = JobRunArgs::with_string("storage", name.clone()); tokio::spawn(async move { registry.trigger(BACKEND_ROTATE_JOB_NAME, &args).await; }); @@ -2625,11 +2622,23 @@ pub async fn list_jobs(State(state): State>) -> impl IntoResponse .iter() .find(|s| s.name == job.name) { + // Config keeps these untyped; type them against the same + // declaration the job dispatches under. A parse failure is + // unreachable — `di.rs` panics at boot on exactly this input — + // so an empty map here means the config changed under a + // running server, and showing no parameters beats inventing + // them. + let declared = job.parameters; job.startup = Some(crate::infrastructure::scheduler::StartupTrigger { - force: configured.args.force, - deep: configured.args.deep, - repair: configured.args.repair, - storage: configured.args.storage.clone(), + params: crate::infrastructure::scheduler::JobRunArgs::from_declared( + declared, + configured + .raw_params + .iter() + .map(|(k, v)| (k.as_str(), v.as_str())), + ) + .map(|a| a.iter().map(|(k, v)| (k.to_string(), v.clone())).collect()) + .unwrap_or_default(), }); } } @@ -2653,25 +2662,21 @@ pub async fn list_jobs(State(state): State>) -> impl IntoResponse /// and `consistency_batch` which fans out to both). Default `false` /// preserves discovery-only. See `JobRunArgs.repair` for the /// content-safety and race-safety guarantees. -#[derive(serde::Deserialize)] -pub struct TriggerJobQuery { - #[serde(default)] - pub force: bool, - #[serde(default)] - pub deep: bool, - /// Optional named storage entry to scope the run against — used by - /// tenants that respect `JobRunArgs.storage` (currently - /// `backend_migration` for its target; `blobs_consistency` / - /// `backend_consistency` will pick this up in slice 7 to probe a - /// non-active entry). Ignored by tenants that don't declare a - /// semantic for it. Unknown-name validation is per-tenant — the - /// generic trigger endpoint doesn't cross-check against - /// `AppConfig.storage_entries`. - #[serde(default)] - pub storage: Option, - #[serde(default)] - pub repair: bool, -} +/// Free-form trigger parameters, validated against the target job's +/// declaration rather than against a fixed field list. +/// +/// A plain `HashMap`, not a newtype over one: `serde_urlencoded` cannot +/// deserialize a newtype struct at the top level, so wrapping it made +/// axum's `Query` extractor reject **every** trigger with a 400 — even +/// one with no query string at all — before the handler ran. +/// +/// This replaced a struct naming `force` / `deep` / `storage` / +/// `repair`, which meant every job advertised the same four whether it +/// read them or not, and a fifth could not be added without editing it. +/// Now the job says what it accepts and +/// [`JobRunArgs::from_declared`] does the parsing, so an undeclared +/// parameter is a 400 naming the real ones instead of a silent no-op. +pub type TriggerJobQuery = std::collections::HashMap; /// `POST /api/admin/jobs/{name}/trigger` — dispatch one run off-schedule. /// @@ -2701,27 +2706,69 @@ pub async fn trigger_job( axum::extract::Query(query): axum::extract::Query, ) -> impl IntoResponse { use crate::infrastructure::scheduler::JobRunArgs; + + // Parse against the job's own declaration. Unknown job → 404 here + // rather than after dispatch, and an undeclared parameter → 400 + // naming what the job does accept. + // Same body as the dispatch-time 404 below — `error` + `name`. + // Clients switch on `error`, so an early return with different + // wording would make "unknown job" mean two things depending on how + // far the request happened to get. This path only exists because the + // declaration has to be read BEFORE the query can be parsed. + let Some(declared) = state.core.job_registry.parameters_of(&name).await else { + return ( + StatusCode::NOT_FOUND, + Json(serde_json::json!({ + "error": "job not registered", + "name": name, + })), + ) + .into_response(); + }; + let args = match JobRunArgs::from_declared( + declared, + query.iter().map(|(k, v)| (k.as_str(), v.as_str())), + ) { + Ok(a) => a, + Err(reason) => { + // Audited: a rejected trigger is an operator action that did + // not happen, and the panel only shows the message. + tracing::info!( + target: "audit", + event = "job.trigger_rejected", + reason = "bad_parameters", + job = %name, + detail = %reason, + "👮🏻‍♂️ Admin trigger rejected for {name}: {reason}", + ); + return ( + StatusCode::BAD_REQUEST, + Json(serde_json::json!({ "error": reason })), + ) + .into_response(); + } + }; + // Audit line BEFORE dispatch so an operator triggering something - // that then hangs still leaves a trail. + // that then hangs still leaves a trail. Parameters are rendered from + // the parsed args rather than named individually, so a job growing + // one does not need this line edited — and cannot end up triggered + // with something the audit trail never recorded. + let params_desc = if args.is_empty() { + "none".to_string() + } else { + args.iter() + .filter_map(|(k, v)| v.to_param_string().map(|s| format!("{k}={s}"))) + .collect::>() + .join(", ") + }; tracing::info!( target: "audit", event = "job.trigger", job = %name, - force = query.force, - deep = query.deep, - repair = query.repair, - "👮🏻‍♂️ Admin triggered job {} (force={}, deep={}, repair={})", - name, - query.force, - query.deep, - query.repair, + params = %params_desc, + "👮🏻‍♂️ Admin triggered job {name} ({params_desc})", ); - let args = JobRunArgs { - force: query.force, - deep: query.deep, - storage: query.storage.clone(), - repair: query.repair, - }; // Jobs that can run for hours (backend_migration, future // reextract_*) are detached: `tokio::spawn` the trigger so the @@ -3141,3 +3188,44 @@ pub async fn purge_job_runs( Err(e) => AppError::internal_error(format!("purge failed: {e}")).into_response(), } } + +#[cfg(test)] +mod tests { + use super::*; + + /// The job-trigger query must extract from any URL shape, including + /// one with no query string at all. + /// + /// Regression: `TriggerJobQuery` was briefly a newtype over the map + /// (`struct TriggerJobQuery(HashMap<..>)`). `serde_urlencoded` + /// cannot deserialize a newtype struct at the top level, so axum's + /// `Query` extractor rejected EVERY trigger with a 400 before the + /// handler body ran — including bare `POST …/dedup_gc/trigger`. It + /// compiled, and it read as if the free-form parameters had been + /// rejected by validation, which sent the first diagnosis at the + /// wrong layer entirely. + #[test] + fn trigger_query_extracts_from_every_url_shape() { + fn parse(uri: &str) -> TriggerJobQuery { + axum::extract::Query::::try_from_uri(&uri.parse().unwrap()) + .unwrap_or_else(|e| panic!("extractor rejected `{uri}`: {e}")) + .0 + } + + assert!(parse("http://x/api/admin/jobs/dedup_gc/trigger").is_empty()); + assert!(parse("http://x/api/admin/jobs/dedup_gc/trigger?").is_empty()); + + let one = parse("http://x/api/admin/jobs/dedup_gc/trigger?force=true"); + assert_eq!(one.get("force").map(String::as_str), Some("true")); + + let two = parse("http://x/t?deep=true&storage=s3_prod"); + assert_eq!(two.get("deep").map(String::as_str), Some("true")); + assert_eq!(two.get("storage").map(String::as_str), Some("s3_prod")); + + // Undeclared names must reach the handler rather than being + // dropped by the extractor — rejecting them, with a message + // naming the job's real parameters, is the handler's job. + let typo = parse("http://x/t?repare=true"); + assert_eq!(typo.get("repare").map(String::as_str), Some("true")); + } +}