From a4101743e08df7b7a278e147362840e6a3ec6725 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 7 Sep 2026 12:30:40 +0200 Subject: [PATCH 1/6] feat(jobs): jobs declare their own run parameters MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `JobRunArgs` was a fixed struct — `force`, `deep`, `storage`, `repair` — 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, the panel's checkboxes, and `StartupTrigger` on the wire. Two costs. Adding a parameter meant editing all six, and forgetting one dropped it silently — most damagingly in persist/restore, where a resumed run lost it and a `?repair=true` migration came back as discovery-only after a restart. And the panel offered the same knobs on every job: only two jobs read `deep`, six read `repair`, so most of those controls did nothing with no way to tell which. Now `JobHandler::parameters()` returns `&'static [JobParam]` — name, type (boolean/string/number), default, and the job's own description of what it does. `JobRunArgs` holds a map keyed by those names. Everything reads the declaration: * `run_or_resume` iterates it to persist and restore, replacing `const FLAGS` plus a `storage` special case. `storage` stops being special — it was the one Option among three bools. * `dispatch` normalises every run against it, which is what makes "a handler sees its declared parameters with their declared defaults" true rather than usual. The periodic tick passes an empty `JobRunArgs::default()`, so a `default: true` parameter would otherwise read false on every scheduled run. * The trigger endpoint takes free-form query params and rejects undeclared ones with a 400 naming the real set, instead of ignoring them. * OXICLOUD_STARTUP_JOBS keeps raw pairs (config is parsed before the registry exists) and validates at dispatch, where the error can name the job's actual parameters. Still a boot panic, same as an unknown job name — a typo'd `?repare=true` must not leave a migration importing forever in discovery mode. * `JobSummary.parameters` carries it to the panel, whose `supportsDeep` was a hardcoded name allowlist (`consistency_batch || backend_consistency`). A job gaining a deep mode needed a frontend release; one losing it left a button that silently did nothing. The menu now renders from the declaration, so a newly-declared boolean appears with no frontend change. Three consistency tenants were hand-rolling persist-on-fresh / restore-on-resume for their own flag, under the same `params` key the engine already used. Deleted — they read `args.get_bool(…)` now. Fresh runs also filter to the declaration. `consistency_batch` forwards its args verbatim to sub-jobs, so a tenant's `params` row could grow `deep` with no deep mode, and the run-detail view would claim a mode the job never had. Two things found while wiring it, both worth knowing: `RecoverableAdapter` bridges the two traits, and `parameters` has to be forwarded there or the registry sees `&[]`. Both traits have defaults, so omitting it compiled cleanly — and the trigger endpoint then rejected `?repair=true` on the very jobs that declare it, with OXICLOUD_STARTUP_JOBS panicking at boot. Now covered by `adapter_forwards_job_metadata_from_inner_handler`. `TriggerJobQuery` was briefly a newtype over the map. `serde_urlencoded` cannot deserialize a newtype struct at the top level, so axum's `Query` rejected EVERY trigger with a 400 — even one with no query string — before the handler ran. It reads exactly like the new validation rejecting something, which sent the first diagnosis to the wrong layer. Now covered by `trigger_query_extracts_from_every_url_shape`. Wire names are a compatibility surface: `params` rows are keyed by them and the panel switches on them, so a rename breaks existing run history the same way renaming a `Mutates` variant does. The JSON shape is pinned in `snapshot_carries_job_metadata`. Co-Authored-By: Claude Opus 5 (1M context) --- frontend/src/lib/api/endpoints/adminJobs.ts | 55 +- frontend/src/lib/api/types.ts | 45 +- .../src/lib/components/AdminJobsPanel.svelte | 107 ++-- src/common/config.rs | 115 ++-- src/common/di.rs | 53 +- src/infrastructure/scheduler/engine.rs | 8 +- src/infrastructure/scheduler/handler.rs | 17 +- src/infrastructure/scheduler/mod.rs | 5 +- src/infrastructure/scheduler/recoverable.rs | 138 +++-- src/infrastructure/scheduler/registry.rs | 82 ++- src/infrastructure/scheduler/types.rs | 491 ++++++++++++++++-- .../services/backend_consistency_service.rs | 99 ++-- .../services/backend_migration_service.rs | 17 +- .../services/backend_rotate_service.rs | 13 +- .../services/blobs_consistency_service.rs | 54 +- .../services/consistency_batch_service.rs | 32 +- src/infrastructure/services/dedup_service.rs | 21 +- .../services/grant_cleanup_service.rs | 20 +- .../services/manifests_consistency_service.rs | 48 +- .../services/thumb_attached_import_service.rs | 13 +- .../services/thumb_derived_import_service.rs | 14 +- .../services/transcode_import_service.rs | 14 +- src/interfaces/api/handlers/admin_handler.rs | 178 +++++-- 23 files changed, 1242 insertions(+), 397 deletions(-) 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")); + } +} From 51e3d614b21930c1ef491bb9d882802f99316173 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 7 Sep 2026 12:41:39 +0200 Subject: [PATCH 2/6] fix(consistency): deep mode must verify storage, not the cache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `backend_consistency ?deep=true` re-reads every chunk and re-hashes it to catch silent bit-rot, and records `blob_corrupted` (severity `data_loss`) naming `backend.backend_type()`. It was reading through the live backend — which for a remote backend includes `CachedBlobBackend`, whose `get_blob_stream` returns the local cached file and never touches the remote on a hit. So the attribution was false in both directions: rot on S3 hidden by a good cached copy, and rot in the cache reported against a healthy S3 — the second sending an operator to the wrong layer entirely. Surfaced by a real run: 2022 chunks, 321 ms shallow, 1.5 s deep. That is 0.74 ms per chunk for a full read plus BLAKE3, sequential, over S3 — impossible, and explained by every chunk being cache-warm. A genuine uncached sweep is tens of seconds. Adds `BlobStorageBackend::uncached()`, defaulting to `None`. `CachedBlobBackend` returns its inner; `Retry` and `Swappable` forward so the unwrap reaches the cache through them. `Swappable` resolves via `current()` rather than capturing a handle, because it sits OUTSIDE the cache — a DI-time snapshot would keep pointing at pre-cutover storage and audit the backend a migration just moved away from. Only the cache is peeled. The cache stores plaintext and the content hash is over plaintext, so unwrapping past the encryption decorator would hand back ciphertext and fail every blob it checked. **No change to normal reads.** `uncached()` is called in exactly one place, and the unwrapped handle is used at exactly one call site (`verify_bytes`). Enumeration, every other job, and every request path still go through the cached stack. `?storage=` was already correct — `build_entry_backend` has no cache decorator — so this only fixes the live-backend path, which is the one that was silently fast. Also reports `verified` in the run extras, on every run including zero. A deep run that verified nothing and one that verified everything were otherwise indistinguishable in the outcome, which is what made a 1.5 s "deep" sweep look plausible in the first place. Same lesson as `orphans_covered` on the Azure fallback: a check that cannot report its own coverage will eventually be believed when it should not be. Co-Authored-By: Claude Opus 5 (1M context) --- src/application/ports/blob_storage_ports.rs | 28 ++++++++++++ .../services/backend_consistency_service.rs | 44 +++++++++++++++++-- .../services/cached_blob_backend.rs | 8 ++++ .../services/retry_blob_backend.rs | 8 ++++ .../services/swappable_blob_backend.rs | 9 ++++ 5 files changed, 93 insertions(+), 4 deletions(-) diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index ae8732d3..7f62d750 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -220,6 +220,34 @@ pub trait BlobStorageBackend: Send + Sync + 'static { /// return `None`; callers that need a local file must stream + spool. fn local_blob_path(&self, hash: &str) -> Option; + /// The same storage with any read-through cache peeled off, or + /// `None` when this backend is not a cache. + /// + /// **For verification only** — normal reads must keep going through + /// the cache, which is the point of having one. + /// + /// A cache answers reads from its own copy, so re-hashing through + /// one checks the cache rather than storage: rot on the remote is + /// hidden by a good cached copy, and rot in the cache is blamed on a + /// healthy remote. The second is worse, because it sends an operator + /// to the wrong layer. `backend_consistency ?deep=true` is the only + /// caller. + /// + /// Peels **only** the cache. The cache sits outside the encryption + /// decorator and stores plaintext, while the content hash is over + /// plaintext, so unwrapping further would hand back ciphertext and + /// fail every blob it checked. + /// + /// Implement by returning the inner backend. Pass-through wrappers + /// (hot-swap, retry) forward to whatever they wrap, so the unwrap + /// still reaches the cache — and, for hot-swap, resolves through + /// `current()` so it survives a migration cutover rather than + /// pinning the pre-cutover storage. Everything else inherits the + /// `None` default and is used as-is. + fn uncached(&self) -> Option> { + None + } + /// How many chunk fetches the CDC reader may run concurrently when /// reassembling a file (`read_blob_stream`'s `buffered(N)` read-ahead). /// diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 38204047..01cd25ac 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -352,11 +352,33 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // actually verified. let deep = args.get_bool("deep"); + // Verify through storage, never through a read-through cache. + // + // A cache answers from its own copy, so re-hashing through one + // checks the CACHE: rot on the remote is masked by a good cached + // copy, and rot in the cache is recorded as `blob_corrupted` + // against a healthy remote — sending an operator to the wrong + // layer. The finding names `backend.backend_type()`, so that + // attribution has to be true. + // + // Only the cache is peeled; the decryptor stays, because the + // cache holds plaintext and the content hash is over plaintext. + // `?storage=` already builds an uncached stack, so this + // only changes the live-backend path — which is the one that was + // silently fast. + let verify_backend = backend.uncached().unwrap_or_else(|| backend.clone()); + // Counter, not just a flag: a deep run that verified nothing and + // a deep run that verified everything are otherwise + // indistinguishable in the outcome, which is exactly the + // ambiguity that made a 1.5s "deep" sweep over 2022 chunks look + // plausible. + let mut verified_count = 0u64; if deep { tracing::info!( target: "oxicloud::consistency", event = "backend_consistency.deep_mode_active", run_id = %store.run_id(), + cached_read_bypassed = backend.uncached().is_some(), "deep mode: re-reading + re-hashing every matched blob (bit-rot detection)" ); } @@ -477,10 +499,14 @@ impl RecoverableJobHandler for BackendConsistencyCheck { event = "backend_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, + verified = verified_count, "backend_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::completed(); + return RunOutcome::completed_with(serde_json::json!({ + "deep": deep, + "verified": verified_count, + })); } // ── Merge-join, not a one-sided probe ──────────────── @@ -562,8 +588,10 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // verified then. (Some(b), Some(d)) if b.hash == **d => { if deep && in_range(&b.hash) { - finding_count += - self.verify_bytes(store, backend.as_ref(), &b.hash).await; + verified_count += 1; + finding_count += self + .verify_bytes(store, verify_backend.as_ref(), &b.hash) + .await; } bi.next(); di.next(); @@ -668,10 +696,18 @@ impl RecoverableJobHandler for BackendConsistencyCheck { event = "backend_consistency.completed", run_id = %store.run_id(), finding_count = finding_count, + verified = verified_count, "backend_consistency completed with {} finding(s)", finding_count ); - return RunOutcome::completed(); + // `verified` is reported on EVERY run, zero included: + // absent-vs-zero is exactly the distinction an operator + // needs, and omitting it on a shallow run would make + // "deep verified nothing" look like "this was shallow". + return RunOutcome::completed_with(serde_json::json!({ + "deep": deep, + "verified": verified_count, + })); } } } diff --git a/src/infrastructure/services/cached_blob_backend.rs b/src/infrastructure/services/cached_blob_backend.rs index b9ee0d95..87f23819 100644 --- a/src/infrastructure/services/cached_blob_backend.rs +++ b/src/infrastructure/services/cached_blob_backend.rs @@ -413,6 +413,14 @@ impl BlobStorageBackend for CachedBlobBackend { if path.exists() { Some(path) } else { None } } + /// This decorator IS the cache, so peeling it yields the real + /// storage. See [`BlobStorageBackend::uncached`] for why an + /// integrity check must not read through here — every other caller + /// keeps using the cache. + fn uncached(&self) -> Option> { + Some(self.inner.clone()) + } + /// Enumeration MUST delegate to the primary (inner) backend, not /// the local cache. The cache is by definition a subset (only /// recently-accessed blobs); walking the cache would look like diff --git a/src/infrastructure/services/retry_blob_backend.rs b/src/infrastructure/services/retry_blob_backend.rs index e06dcc30..0141a36f 100644 --- a/src/infrastructure/services/retry_blob_backend.rs +++ b/src/infrastructure/services/retry_blob_backend.rs @@ -356,6 +356,14 @@ impl BlobStorageBackend for RetryBlobBackend { self.inner.local_blob_path(hash) } + /// Pass-through wrapper — forward, so an unwrap started above still + /// reaches the cache. Inheriting the `None` default would silently + /// end the search at this layer and leave verification reading + /// through the cache after all. + fn uncached(&self) -> Option> { + self.inner.uncached() + } + /// Enumeration delegates to inner. Retry semantics apply per /// call, not per batch — a single list call that fails after /// exhausting retries surfaces the error to the tenant, which diff --git a/src/infrastructure/services/swappable_blob_backend.rs b/src/infrastructure/services/swappable_blob_backend.rs index e28abadb..8e8926c4 100644 --- a/src/infrastructure/services/swappable_blob_backend.rs +++ b/src/infrastructure/services/swappable_blob_backend.rs @@ -205,6 +205,15 @@ impl BlobStorageBackend for SwappableBlobBackend { self.current().local_blob_path(hash) } + /// Resolved through `current()`, not captured once: this wrapper sits + /// OUTSIDE the cache, so a migration cutover replaces the entire + /// cached stack beneath it. A handle taken at DI time would keep + /// pointing at the pre-cutover storage and audit the backend that + /// was just migrated away from. + fn uncached(&self) -> Option> { + self.current().uncached() + } + fn read_prefetch(&self) -> usize { self.current().read_prefetch() } From 9514f49eed04ffcbdca30ed50cefbe9fd60d5254 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 7 Sep 2026 13:13:17 +0200 Subject: [PATCH 3/6] feat(consistency): record what a deep run audited, and make cancel responsive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two problems a real S3 run exposed, both about a completed run being unable to answer questions about itself. ## Which storage did this verify? A finished run recorded `deep`, `verified`, `total_rows` — but not its target. Findings carry `"backend"`, and a clean run has none, so a green audit says nothing about what it audited. After switching the active backend there is no way to tell what a previous run covered. That is not hypothetical: a 1.5 s local sweep was read as an S3 audit by both of us for several exchanges, and the run JSON could not settle it. What settled it was the ABSENCE of a `storage` param, inferred by hand. Now the outcome carries `backend` (the type), `storage_entry` (the entry name) and `scoped` (whether `?storage=` was given). The entry name is read from `admin_settings` at run start rather than snapshotted at boot, because a migration cutover rewrites it while the process lives — a cached copy would name the pre-cutover entry, which is the same staleness trap the `uncached()` unwrap avoids by resolving through `current()`. Best-effort: it is a label, and failing an audit over one would be the wrong trade. `ActiveEntry::Unset` stays unlabelled rather than guessing at the boot fallback. ## Cancel was bounded by a batch, and a batch got 60,000x slower `BATCH_SIZE`'s comment claimed 500 "keeps the cancel-poll cadence sub-second (each batch = one backend list + one DB probe + Rust set-difference)". True when written. Deep mode then moved into this tenant and added 500 full blob reads per batch: measured at 155 ms each against OVH S3, so ~63 s per batch. The status poll ran only between batches, so Pause and Cancel appeared ignored for a minute — on exactly the run an operator most wants to stop, and one that scales to hours on a real corpus. Cancellation is now polled inside the verify loop every `DEEP_CANCEL_POLL_EVERY` (16) blobs. A poll is one indexed DB read (~0.1 ms) against a 155 ms remote read, so the cost is under 1% there and a couple of percent even on a local backend where a verify is ~0.8 ms. `BATCH_SIZE` goes back to being purely about I/O batching, and its comment now says so. Pausing mid-batch is safe: the cursor still points at the last completed batch, so a resume re-verifies this batch's handful of blobs rather than skipping them. Re-reading a few is the right direction for a check whose whole purpose is not missing anything. Measurements quoted throughout are from a live run: 2022 chunks (231 files), 314 s against remote S3 versus 1.567 s local, verified 2022 in both. Co-Authored-By: Claude Opus 5 (1M context) --- .../services/backend_consistency_service.rs | 109 ++++++++++++++++-- 1 file changed, 99 insertions(+), 10 deletions(-) diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 01cd25ac..e039d508 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -124,14 +124,37 @@ pub const BACKEND_CONSISTENCY_JOB_NAME: &str = "backend_consistency"; /// touches a backend and so has no entry to scope. pub const PROBED_STORAGE_PARAM: &str = "probed_storage"; -/// Batch size for backend enumeration + DB probe. 500 is enough to -/// amortise the DB round-trip while keeping the cancel-poll cadence -/// sub-second (each batch = one backend list + one DB probe + Rust -/// set-difference). Larger batches on S3 hit ListObjectsV2's -/// per-request limit (1000) with wasted rows filtered client-side; -/// smaller batches over-poll the DB. +/// Batch size for backend enumeration + DB probe. 500 amortises the DB +/// round-trip; larger batches on S3 hit ListObjectsV2's per-request +/// limit (1000) with wasted rows filtered client-side, and smaller ones +/// over-poll the DB. +/// +/// **This is not the cancel cadence.** It used to be: a shallow batch is +/// one backend list + one DB probe + a Rust set-difference, so polling +/// once per batch kept cancel sub-second. Deep mode then moved into this +/// tenant and added 500 full blob reads per batch — measured at 155 ms +/// each against remote S3, so ~63 s per batch — and a cancel poll that +/// only ran between batches left Pause/Cancel unresponsive for a minute +/// on exactly the run an operator most wants to stop. +/// +/// [`DEEP_CANCEL_POLL_EVERY`] decouples the two: cancellation is now +/// checked inside the verify loop, so this constant went back to being +/// purely about I/O batching. const BATCH_SIZE: usize = 500; +/// How many deep verifications to run between cancel polls. +/// +/// A poll is one small indexed DB read (~0.1 ms) against a blob read +/// measured at 155 ms on remote storage, so polling every blob would +/// cost well under 1%. 16 keeps even a local-backend deep run — where a +/// verify is ~0.8 ms and the ratio is far less favourable — under a +/// couple of percent, while capping cancel latency at well under a +/// second on any backend. +/// +/// Shallow batches do not need this: they are already fast enough that +/// the per-batch poll bounds latency on its own. +const DEEP_CANCEL_POLL_EVERY: u64 = 16; + /// Grace window — orphans younger than this are skipped, since the /// write path is durability-before-visibility: bytes hit disk before /// the `storage.blobs` row is inserted. A scan catching a blob @@ -294,6 +317,40 @@ impl RecoverableJobHandler for BackendConsistencyCheck { } None => None, }; + // The entry name this run audits, recorded in the outcome so a + // finished run says WHAT it checked. + // + // Without it a completed run is silent about its target: findings + // carry `backend`, but a clean run has none, so after switching + // the active backend there is no way to tell which storage a + // previous green run actually verified. Ed hit exactly that — + // a 1.5s local sweep read as an S3 audit. + // + // Read from `admin_settings` rather than a boot-time snapshot, + // because a migration cutover rewrites it while the process + // lives; a cached copy would name the pre-cutover entry. Best + // effort: this is a label, and failing an audit over it would be + // the wrong trade. + let audited_entry: Option = match &probed_storage { + Some(name) => Some(name.clone()), + None => { + match crate::infrastructure::services::entry_backend::resolve_active_entry( + self.pool.as_ref(), + &self.storage_entries, + ) + .await + { + Ok(crate::infrastructure::services::entry_backend::ActiveEntry::Explicit( + e, + )) => Some(e.name.clone()), + // Unset means the boot fallback picked the first + // entry; naming it would be a guess, and a wrong + // label is worse than an absent one. + _ => None, + } + } + }; + let backend: Arc = match &probed_storage { None => self.backend.clone(), Some(name) => match self.storage_entries.iter().find(|e| &e.name == name) { @@ -504,8 +561,11 @@ impl RecoverableJobHandler for BackendConsistencyCheck { finding_count ); return RunOutcome::completed_with(serde_json::json!({ - "deep": deep, - "verified": verified_count, + "deep": deep, + "verified": verified_count, + "backend": backend.backend_type(), + "storage_entry": audited_entry, + "scoped": probed_storage.is_some(), })); } @@ -588,6 +648,32 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // verified then. (Some(b), Some(d)) if b.hash == **d => { if deep && in_range(&b.hash) { + // Cancel poll INSIDE the verify loop. The + // per-batch poll above bounds latency by one + // batch, which is ~63 s of remote reads in + // deep mode — a minute of an apparently + // ignored Cancel on a run that may last + // hours. Pausing here is safe and cheap: the + // cursor still points at the last completed + // batch, so resume re-verifies this batch's + // handful of blobs rather than skipping them. + if verified_count.is_multiple_of(DEEP_CANCEL_POLL_EVERY) + && matches!(store.status().await, Ok(RunStatus::CancelRequested)) + { + tracing::info!( + target: "oxicloud::consistency", + event = "backend_consistency.cancelled_mid_verify", + run_id = %store.run_id(), + verified = verified_count, + "deep verify cancelled mid-batch, pausing at the last checkpoint" + ); + return RunOutcome::Paused { + cursor: cursor + .as_ref() + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(), + }; + } verified_count += 1; finding_count += self .verify_bytes(store, verify_backend.as_ref(), &b.hash) @@ -705,8 +791,11 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // needs, and omitting it on a shallow run would make // "deep verified nothing" look like "this was shallow". return RunOutcome::completed_with(serde_json::json!({ - "deep": deep, - "verified": verified_count, + "deep": deep, + "verified": verified_count, + "backend": backend.backend_type(), + "storage_entry": audited_entry, + "scoped": probed_storage.is_some(), })); } } From fc88a78055970caae82717376188ac9f2802c48a Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 7 Sep 2026 13:19:18 +0200 Subject: [PATCH 4/6] fix(consistency): a mid-batch pause resumes at the last settled hash MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `04807464` made deep-mode cancellation responsive but paused at the BATCH-START cursor, which throws away everything done in the current batch. Ed caught the sharp edge while testing pause: the checkpoint only lands after a batch completes, so a run paused 27 s into its FIRST 63 s batch had `cursor_hex: ""` and would resume from scratch. Later batches lose 500 blob reads, about a minute against remote S3. Now the pause carries `settled` — the highest hash whose pair was fully handled. That is safe because the merge-join advances both sides in ascending hash order: at any point in the loop, everything at or below `settled` has had its findings recorded and, under `?deep=true`, its bytes re-hashed. So resume re-does one pair, not the whole batch. `settled` only advances after an arm finishes with its item, never on entry, which is what keeps that invariant true. Also checkpoints explicitly before returning `Paused`, with `delta_count = 0` since `scanned_count` is already credited per batch. The engine writes the cursor on the Paused row anyway; persisting it here means a restart racing that write still resumes from the right place rather than the previous batch. Strictly fewer duplicate findings on resume, too. Page-level `unknown_backend_file` notices are emitted before the join, so any resume re-emits those for the re-walked range — a shorter range is simply less of it. That duplication is pre-existing and orthogonal. Co-Authored-By: Claude Opus 5 (1M context) --- .../services/backend_consistency_service.rs | 64 ++++++++++++++----- 1 file changed, 48 insertions(+), 16 deletions(-) diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index e039d508..6107e193 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -635,6 +635,11 @@ impl RecoverableJobHandler for BackendConsistencyCheck { let mut bi = page.blobs.iter().peekable(); let mut di = db_hashes.iter().peekable(); + // Highest hash fully handled in THIS batch — findings + // recorded, bytes verified if deep. A mid-batch pause resumes + // here, so it must only advance once an arm is finished with + // its item, never on entry. + let mut settled: Option = None; loop { match (bi.peek(), di.peek()) { // Present on both sides. Shallow: nothing to say — the @@ -648,37 +653,62 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // verified then. (Some(b), Some(d)) if b.hash == **d => { if deep && in_range(&b.hash) { - // Cancel poll INSIDE the verify loop. The - // per-batch poll above bounds latency by one - // batch, which is ~63 s of remote reads in - // deep mode — a minute of an apparently - // ignored Cancel on a run that may last - // hours. Pausing here is safe and cheap: the - // cursor still points at the last completed - // batch, so resume re-verifies this batch's - // handful of blobs rather than skipping them. + // Cancel poll INSIDE the verify loop, because + // the per-batch poll bounds latency by one + // batch — ~63 s of remote reads in deep mode, + // a minute of an apparently ignored Cancel on + // a run that may last hours. + // + // Pauses at `settled`, the last pair fully + // handled, NOT at the batch's start cursor. + // The merge-join walks both sides in + // ascending hash order, so everything at or + // below `settled` has had its findings + // recorded and its bytes verified — resuming + // there re-does one pair, not five hundred. + // + // The batch-start cursor would have been + // correct but wasteful: in the FIRST batch it + // is empty, so a pause 27 s into a 63 s batch + // threw away the whole scan. if verified_count.is_multiple_of(DEEP_CANCEL_POLL_EVERY) && matches!(store.status().await, Ok(RunStatus::CancelRequested)) { + let resume_at = settled.clone().or_else(|| cursor.clone()); tracing::info!( target: "oxicloud::consistency", event = "backend_consistency.cancelled_mid_verify", run_id = %store.run_id(), verified = verified_count, - "deep verify cancelled mid-batch, pausing at the last checkpoint" + resume_at = resume_at.as_deref().unwrap_or(""), + "deep verify cancelled mid-batch, pausing at the last settled hash" ); - return RunOutcome::Paused { - cursor: cursor - .as_ref() - .map(|s| s.as_bytes().to_vec()) - .unwrap_or_default(), - }; + // Checkpoint so the cursor survives even + // if the engine's Paused write races a + // restart; `scanned_count` is already + // counted per batch, so add nothing here. + let bytes = resume_at + .as_ref() + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(); + if let Err(e) = store.checkpoint(bytes.clone(), 0).await { + tracing::warn!( + target: "oxicloud::consistency", + event = "backend_consistency.pause_checkpoint_failed", + run_id = %store.run_id(), + error = %e, + "could not persist the mid-batch cursor; resume will \ + restart from the previous batch" + ); + } + return RunOutcome::Paused { cursor: bytes }; } verified_count += 1; finding_count += self .verify_bytes(store, verify_backend.as_ref(), &b.hash) .await; } + settled = Some(b.hash.clone()); bi.next(); di.next(); } @@ -707,6 +737,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { ) .await; } + settled = Some(b.hash.clone()); bi.next(); } // DB-only: a row whose bytes are gone. Severity is @@ -729,6 +760,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { }), ) .await; + settled = Some((*d).clone()); di.next(); } // Past the horizon on both sides, or both exhausted. From 3da8cd663db1c365b20cfaf5f5f674b5065a817d Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 7 Sep 2026 17:58:29 +0200 Subject: [PATCH 5/6] ci: retrigger ci From 5b3d5cbb10e074e52b5a87a200d041b081e9eab6 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Mon, 7 Sep 2026 22:48:40 +0200 Subject: [PATCH 6/6] CI: retrigger CI src/lib/api/endpoints/recipients.bench.test.ts is too sensitive and generates false positive on loaded worker