From 6b7bb67500070503a07ae69557d5a4f834ebbe5b Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sat, 1 Aug 2026 12:59:35 +0200 Subject: [PATCH] feat(storage): jobs can choose storage to migrate/scan --- src/application/dtos/settings_dto.rs | 17 ++ src/common/di.rs | 36 ++- src/infrastructure/scheduler/pg_job_store.rs | 31 +++ src/infrastructure/scheduler/recoverable.rs | 32 +++ src/infrastructure/scheduler/types.rs | 12 + .../services/storage_migration_service.rs | 248 +++++++++++++++--- src/interfaces/api/handlers/admin_handler.rs | 58 +++- 7 files changed, 385 insertions(+), 49 deletions(-) diff --git a/src/application/dtos/settings_dto.rs b/src/application/dtos/settings_dto.rs index 05f35b6f..fe04d4b2 100644 --- a/src/application/dtos/settings_dto.rs +++ b/src/application/dtos/settings_dto.rs @@ -257,9 +257,26 @@ pub struct MigrationStateDto { } /// Request body for `POST /api/admin/storage/migration/start`. +/// +/// **Multi-entry contract** (see `docs/plan/storage-multi-entry.md`): +/// `target_name` is REQUIRED — it names the storage entry the copy +/// job will move blobs INTO. The admin picks it from the entries +/// declared in `OXICLOUD_STORAGE_ENTRIES`. The trigger endpoint +/// rejects the request when the name doesn't exist or equals the +/// currently-active entry (no-op guard). #[derive(Debug, Serialize, Deserialize, ToSchema)] pub struct StartMigrationDto { + /// Name of the storage entry to migrate blobs INTO. Must be + /// present in `OXICLOUD_STORAGE_ENTRIES` and must differ from + /// the currently-active entry. + pub target_name: String, /// How many blobs to copy in parallel (default: 4). + /// + /// **Currently ignored** — the recoverable copy loop is + /// sequential (one blob at a time within the batch). Kept in + /// the DTO for wire-compat with the admin UI form; will be + /// honoured once per-batch fan-out lands (dual-write / + /// concurrent-copy future slice). pub concurrency: Option, } diff --git a/src/common/di.rs b/src/common/di.rs index 9edecebb..540c8119 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -237,8 +237,16 @@ impl AppServiceFactory { // working. Encryption never applies here (legacy synthesis // would have created an entry if any legacy var was set). let active_backend_kind: StorageBackendType; + // Track which named entry the LIVE backend was built from so + // the migration handler can name-compare `target != active` + // without re-reading the DB. `"legacy"` sentinel for the + // no-entries branch — the migration handler refuses that + // target name anyway (no entry exists), which is the correct + // behaviour for the zero-config path. + let active_backend_name: String; let base_backend: Arc = if self.config.storage_entries.is_empty() { active_backend_kind = self.config.storage.backend.clone(); + active_backend_name = "legacy".to_string(); tracing::info!( "Storage: no OXICLOUD_STORAGE_ENTRIES declared and no legacy vars — using \ framework default (backend={:?}, path={:?})", @@ -312,6 +320,7 @@ impl AppServiceFactory { } }; active_backend_kind = entry.backend.clone(); + active_backend_name = entry.name.clone(); build_entry_backend(entry, &self.storage_path) }; @@ -549,6 +558,7 @@ impl AppServiceFactory { job_registry, job_store_provider, blob_backend: blob_backend_for_consistency, + active_backend_name, }) } @@ -2167,13 +2177,11 @@ impl AppServiceFactory { tracing::info!("Storage settings service initialized"); // 9b-1c. Register the storage-backend migration tenant on - // the recoverable-run engine. Must run AFTER the storage - // settings service is built — the tenant resolves the - // *target* backend at each run start by asking the settings - // service for the currently-effective config. Source is - // whatever `dedup_service` booted with; both live on - // `AppState.core`. On-demand only (no periodic tick — an - // operator triggers a copy after switching backend config). + // the recoverable-run engine. Target is resolved by NAME + // from `params.target_name` on each run — plumbed from + // the trigger endpoint. Constructor takes the ambient + // entries snapshot + active-name + storage_path fallback + // so no DB read is needed per run for target lookup. let job_store_provider_dyn: Arc< dyn crate::infrastructure::scheduler::JobStoreProvider, > = app_state.core.job_store_provider.clone(); @@ -2184,7 +2192,9 @@ impl AppServiceFactory { .clone() .expect("maintenance_pool set above"), app_state.core.blob_backend.clone(), - storage_settings_svc, + app_state.core.active_backend_name.clone(), + app_state.core.config.storage_entries.clone(), + self.storage_path.clone(), ), ) .register_recoverable_job(&app_state.core.job_registry, &job_store_provider_dyn) @@ -2436,6 +2446,16 @@ pub struct CoreServices { /// `build_app_state` — can probe `blob_exists()` / re-hash bytes /// through the same stack DedupService uses. pub blob_backend: Arc, + /// Name of the storage entry the LIVE `blob_backend` was built + /// from. Populated at boot: either from + /// `admin_settings.storage.active_backend_name` when set, or the + /// first entry in `OXICLOUD_STORAGE_ENTRIES` when unset. For the + /// no-entries legacy path this is `"default"` (the synthesized + /// name) or `"legacy"` (framework-defaults case with zero storage + /// config at all). Migration handler consumes this to enforce the + /// "target != active" no-op guard by name; without needing to + /// re-read DB on every trigger. + pub active_backend_name: String, } /// Container for repository services diff --git a/src/infrastructure/scheduler/pg_job_store.rs b/src/infrastructure/scheduler/pg_job_store.rs index 64762458..a8a3436c 100644 --- a/src/infrastructure/scheduler/pg_job_store.rs +++ b/src/infrastructure/scheduler/pg_job_store.rs @@ -198,6 +198,37 @@ impl JobStore for PgJobStore { Ok(()) } + async fn set_string_param(&self, key: &str, value: &str) -> Result<(), DomainError> { + sqlx::query( + r#" + UPDATE jobs.recoverable_runs + SET params = jsonb_set(params, ARRAY[$2], to_jsonb($3::text)) + WHERE id = $1 + "#, + ) + .bind(self.run_id) + .bind(key) + .bind(value) + .execute(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("set_string_param", e))?; + Ok(()) + } + + async fn get_string_param(&self, key: &str) -> Result, DomainError> { + // `params -> $2` returns JSONB; `->>` returns text (null when + // key absent OR value isn't a string). Handler treats absence + // and null-value identically — either way it's "not set." + let row: Option<(Option,)> = + sqlx::query_as("SELECT params ->> $2 FROM jobs.recoverable_runs WHERE id = $1") + .bind(self.run_id) + .bind(key) + .fetch_optional(self.pool.as_ref()) + .await + .map_err(|e| map_sqlx_err("get_string_param", e))?; + Ok(row.and_then(|(v,)| v)) + } + async fn mark_completed(&self) -> Result<(), DomainError> { sqlx::query( r#" diff --git a/src/infrastructure/scheduler/recoverable.rs b/src/infrastructure/scheduler/recoverable.rs index eef85e55..dd1a44c5 100644 --- a/src/infrastructure/scheduler/recoverable.rs +++ b/src/infrastructure/scheduler/recoverable.rs @@ -246,6 +246,24 @@ pub trait JobStore: Send + Sync { async fn seed_progress_params(&self, total: u64, kind: ProgressKind) -> Result<(), DomainError>; + /// Set an arbitrary string field on `params` (JSONB). Used by + /// handlers on a Fresh run to persist per-run configuration that + /// must survive a mid-run restart — e.g. `storage_migration` + /// stamping `params.target_name` at run start so a resume can + /// pick up the same target without the admin re-specifying it. + /// + /// Handler-callable (unlike `seed_progress_params`, which is + /// engine-only). Idempotent: re-writing the same value is a + /// no-op UPDATE. + async fn set_string_param(&self, key: &str, value: &str) -> Result<(), DomainError>; + + /// Read a string field from `params` (JSONB). Returns `None` when + /// the key is absent or its value isn't a JSON string. Paired + /// with [`Self::set_string_param`] — handlers on a Resumed run + /// use this to recover per-run config that a prior Fresh open + /// stamped. + async fn get_string_param(&self, key: &str) -> Result, DomainError>; + /// Persist one finding to `jobs.run_findings` and bump /// `stats.finding_count` on the parent run. Consistency handlers /// call this in place of the transitional @@ -838,6 +856,7 @@ mod tests { findings: Vec, progress_total: Option, progress_kind: Option, + string_params: std::collections::HashMap, } #[async_trait] @@ -886,6 +905,17 @@ mod tests { s.progress_kind = Some(kind); Ok(()) } + async fn set_string_param(&self, key: &str, value: &str) -> Result<(), DomainError> { + self.state + .lock() + .unwrap() + .string_params + .insert(key.to_string(), value.to_string()); + Ok(()) + } + async fn get_string_param(&self, key: &str) -> Result, DomainError> { + Ok(self.state.lock().unwrap().string_params.get(key).cloned()) + } async fn mark_completed(&self) -> Result<(), DomainError> { self.state.lock().unwrap().status = RunStatus::Completed; Ok(()) @@ -937,6 +967,7 @@ mod tests { findings: Vec::new(), progress_total: None, progress_kind: None, + string_params: std::collections::HashMap::new(), }), }); let id = store.run_id; @@ -995,6 +1026,7 @@ mod tests { findings: Vec::new(), progress_total: None, progress_kind: None, + string_params: std::collections::HashMap::new(), }), }); stores.push(store.clone()); diff --git a/src/infrastructure/scheduler/types.rs b/src/infrastructure/scheduler/types.rs index 396686c3..7a3c8a9a 100644 --- a/src/infrastructure/scheduler/types.rs +++ b/src/infrastructure/scheduler/types.rs @@ -34,10 +34,22 @@ use serde::{Deserialize, Serialize}; /// - `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`): +/// - `storage_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. #[derive(Debug, Clone, Default)] pub struct JobRunArgs { pub force: bool, pub deep: bool, + pub storage: Option, } /// Uniform outcome the supervisor logs and stores for every job dispatch. diff --git a/src/infrastructure/services/storage_migration_service.rs b/src/infrastructure/services/storage_migration_service.rs index 4f8b0979..eb5f6461 100644 --- a/src/infrastructure/services/storage_migration_service.rs +++ b/src/infrastructure/services/storage_migration_service.rs @@ -46,7 +46,7 @@ //! cancel + cursor discipline; the batch loop is I/O-bound anyway. //! Add concurrency later if a real throughput need appears. -use std::path::Path; +use std::path::{Path, PathBuf}; use std::sync::Arc; use async_trait::async_trait; @@ -54,15 +54,24 @@ use futures::StreamExt; use sqlx::PgPool; use crate::application::ports::blob_storage_ports::BlobStorageBackend; -use crate::application::services::storage_settings_service::StorageSettingsService; +use crate::common::config::NamedStorageEntry; use crate::common::errors::DomainError; use crate::infrastructure::scheduler::{ JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, RunStatus, record_or_log, }; +use crate::infrastructure::services::entry_backend::build_entry_backend; pub const STORAGE_MIGRATION_JOB_NAME: &str = "storage_migration"; +/// The `params` JSONB key under which the run's target entry name is +/// stashed at Fresh-open time via `JobStore::set_string_param`. +/// Handlers re-read it on Resume so a paused run survives a restart +/// without the admin re-specifying the target. Exposed publicly so +/// the trigger endpoint's audit lines and the admin UI's run-detail +/// projections read the same constant. +pub const TARGET_NAME_PARAM: &str = "target_name"; + /// Rows per batch. Copies are I/O-bound (source read + target write); /// larger batches amortise fewer SQL round-trips but the checkpoint /// / cancel-poll cadence lengthens. 100 balances the two — one @@ -72,20 +81,39 @@ const BATCH_SIZE: i64 = 100; pub struct StorageMigrationService { pool: Arc, + /// Backend the running app is bound to — the migration COPIES + /// FROM this. Set once at boot and never changes for the + /// process's lifetime (cutover requires a restart, per plan). source: Arc, - storage_settings: Arc, + /// Name of the currently-active entry (i.e. the one `source` + /// corresponds to). Used to refuse a same-name target at run + /// start. Same reasoning as `source` — locked at boot. + active_backend_name: String, + /// All entries declared in env, held as a snapshot for name + /// lookup during migration. Immutable per-deploy — matches + /// `AppConfig.storage_entries`. + storage_entries: Vec, + /// Ambient `AppConfig.storage_path` used as the `root_dir` + /// fallback for a Local target entry that doesn't declare its + /// own `_ROOT_DIR`. Same fallback rule as boot + /// (`build_entry_backend`). + storage_path_fallback: PathBuf, } impl StorageMigrationService { pub fn new( pool: Arc, source: Arc, - storage_settings: Arc, + active_backend_name: String, + storage_entries: Vec, + storage_path_fallback: PathBuf, ) -> Self { Self { pool, source, - storage_settings, + active_backend_name, + storage_entries, + storage_path_fallback, } } @@ -133,47 +161,152 @@ impl RecoverableJobHandler for StorageMigrationService { async fn run_resumable( &self, store: &dyn JobStore, - _args: &JobRunArgs, + args: &JobRunArgs, resume_cursor: Option>, ) -> RunOutcome { - // No-op guard — refuse when the effective (target) config - // points at the same physical storage as the source (boot - // config). Without this, a misclick on an S3 deployment - // issues one HEAD per blob for zero useful work — cheap on - // local, expensive on remote. Same-type-different-location - // migrations (local dir change, S3 bucket change) pass this - // check and proceed normally. - match self.storage_settings.is_source_target_identical().await { - Ok(true) => { - tracing::warn!( - target: "audit", - event = "storage_migration.refused_noop", - run_id = %store.run_id(), - "storage_migration refused: source and target point at the same storage" - ); + // Resolve the target entry NAME. Two paths: + // + // * Fresh run — `args.storage` MUST be Some (the trigger + // endpoint enforces this at the HTTP layer). Handler + // stamps it into `params.target_name` so a mid-run restart + // can resume without re-input. + // * Resumed run — `args.storage` is typically None (admin + // just clicked Run on a Paused row). Handler reads the + // target from `params.target_name` written on the + // original Fresh open. + // + // A Fresh run without `args.storage` is a client bug — refuse + // rather than default to something and quietly copy blobs + // into the wrong entry. + let is_fresh = resume_cursor.is_none(); + let target_name = if is_fresh { + let Some(name) = args.storage.clone() else { return RunOutcome::Failed { message: - "target equals source; change storage settings before triggering a migration" + "storage_migration requires `target_name` on a fresh run — trigger via \ + POST /api/admin/storage/migration/start with `{\"target_name\": \"\"}`." .to_string(), }; - } - Ok(false) => {} - Err(e) => { + }; + if let Err(e) = store.set_string_param(TARGET_NAME_PARAM, &name).await { return RunOutcome::Failed { - message: format!("identity check: {e}"), + message: format!("failed to persist target_name to params: {e}"), }; } + name + } else { + match store.get_string_param(TARGET_NAME_PARAM).await { + Ok(Some(name)) => name, + Ok(None) => { + return RunOutcome::Failed { + message: format!( + "resumed run has no {TARGET_NAME_PARAM} in params — cannot infer \ + target. Likely a Paused row from before the multi-entry migration \ + refactor; cancel + trigger fresh." + ), + }; + } + Err(e) => { + return RunOutcome::Failed { + message: format!("read {TARGET_NAME_PARAM} from params: {e}"), + }; + } + } + }; + + // First-line guard: target name equals the currently-active + // entry. Silent no-op if we let it through — the app would + // walk every blob and skip because `target.blob_exists` is + // trivially true (target = live source). Even on the same + // local disk that's a lot of syscalls for no reason; on S3 + // it costs one HEAD per blob for zero copies. + if target_name == self.active_backend_name { + tracing::warn!( + target: "audit", + event = "storage_migration.refused_noop", + run_id = %store.run_id(), + target_name = %target_name, + active = %self.active_backend_name, + "storage_migration refused: target equals the currently-active entry" + ); + return RunOutcome::Failed { + message: format!( + "target entry `{target_name}` is the currently-active entry — nothing to \ + migrate. Pick a different target." + ), + }; } - // Resolve target at run start. - let target = match self.storage_settings.build_effective_backend().await { - Ok(t) => t, - Err(e) => { + // Look up the target entry by name. + let target_entry = match self.storage_entries.iter().find(|e| e.name == target_name) { + Some(e) => e, + None => { + let available = if self.storage_entries.is_empty() { + "(none)".to_string() + } else { + self.storage_entries + .iter() + .map(|e| e.name.as_str()) + .collect::>() + .join(", ") + }; return RunOutcome::Failed { - message: format!("resolve target backend: {e}"), + message: format!( + "target entry `{target_name}` not found in OXICLOUD_STORAGE_ENTRIES. \ + Available: [{available}]. If the entry was removed from .env since this \ + run started, restore it or cancel this run." + ), }; } }; + let source_entry = self + .storage_entries + .iter() + .find(|e| e.name == self.active_backend_name); + + // Second-line guard: physical-identity check for the + // encryption-differs case. Two entries with different names + // can still point at the same physical bucket (only their + // encryption key differs). That's the "in-place encryption + // rotation" case — refused because reads during migration + // would fail (LIVE backend uses K1, storage is being + // overwritten with K2). See plan §Encryption "Proper + // in-place rotation" for the deferred fix. The compare uses + // `entry_identity` (backend + physical location, EXCLUDING + // encryption key). + if let Some(source) = source_entry + && entry_identity(source) == entry_identity(target_entry) + { + let key_differs = source.encryption_key_base64 != target_entry.encryption_key_base64; + let hint = if key_differs { + " (encryption key differs → this looks like an in-place key rotation; \ + create a new entry pointing at a DIFFERENT bucket / dir, migrate to it, \ + then move back if desired)" + } else { + "" + }; + tracing::warn!( + target: "audit", + event = "storage_migration.refused_same_physical_storage", + run_id = %store.run_id(), + target_name = %target_name, + source_name = %self.active_backend_name, + encryption_differs = key_differs, + "storage_migration refused: named target differs from source but physical storage matches" + ); + return RunOutcome::Failed { + message: format!( + "target entry `{target_name}` names a different entry than the active \ + `{}`, but they point at the same physical storage{hint}.", + self.active_backend_name, + ), + }; + } + + // Build target backend via the shared factory — same code + // path boot uses, so the encryption decorator wrapping is + // uniform. + let target = build_entry_backend(target_entry, &self.storage_path_fallback); if let Err(e) = target.initialize().await { return RunOutcome::Failed { message: format!("target backend init: {e}"), @@ -186,10 +319,13 @@ impl RecoverableJobHandler for StorageMigrationService { target: "audit", event = "storage_migration.run_started", run_id = %store.run_id(), - source = source_kind, - target = target_kind, - resuming = resume_cursor.is_some(), - "storage_migration starting {source_kind} → {target_kind}" + source_name = %self.active_backend_name, + target_name = %target_name, + source_kind = source_kind, + target_kind = target_kind, + resuming = !is_fresh, + "storage_migration starting {} ({source_kind}) → {target_name} ({target_kind})", + self.active_backend_name, ); // Cursor = the last-visited blob hash, UTF-8-encoded. On resume @@ -424,6 +560,48 @@ impl RecoverableJobHandler for StorageMigrationService { } } +/// Physical-storage identity string for a `NamedStorageEntry`. Two +/// entries with the same identity point at the same physical +/// location (same disk dir, same S3 bucket, same Azure container) +/// regardless of encryption key or credentials. Used by the second- +/// line refusal in `run_resumable` to catch in-place encryption +/// rotation attempts (same physical storage, K1 → K2 → reads-during- +/// migration break). See `docs/plan/storage-multi-entry.md` §Encryption. +/// +/// Deliberately excludes: +/// - Encryption key — otherwise same-bucket-different-key would look +/// like a legit migration, hiding the corruption. +/// - Credentials — two entries with different access keys pointing +/// at the same bucket ARE the same physical storage. +/// - Region for S3 — the bucket URI is the primary key; region is a +/// routing hint (though endpoint_url is included since it changes +/// the actual host bytes land on). +fn entry_identity(entry: &NamedStorageEntry) -> String { + use crate::common::config::StorageBackendType; + match entry.backend { + StorageBackendType::Local => { + format!("local:{}", entry.root_dir.as_deref().unwrap_or("")) + } + StorageBackendType::S3 => match entry.s3.as_ref() { + Some(s3) => format!( + "s3:{}/{}:path_style={}", + s3.endpoint_url.as_deref().unwrap_or("aws"), + s3.bucket, + s3.force_path_style, + ), + None => "s3:".to_string(), + }, + StorageBackendType::Azure => match entry.azure.as_ref() { + Some(az) => format!( + "azure:{}/{}", + az.account_name.as_str(), + az.container.as_str() + ), + None => "azure:".to_string(), + }, + } +} + /// Copy one blob: stream source bytes to a temp file, then hand the /// path to `target.put_blob`. The spool-through-disk shape matches /// what the old `migration_job::copy_blob` did — some backends' diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index eb172bda..a0925a97 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -454,9 +454,36 @@ pub async fn get_migration_status( )] pub async fn start_migration( State(state): State>, - Json(_dto): Json, + Json(dto): Json, ) -> Result { - trigger_storage_migration(state).await + // Validate at the HTTP layer (before spawning) so unknown / no-op + // targets get a synchronous 400 response instead of burning a + // failed run row. The handler's own checks are second-line + // defence for the resume path where args aren't repeated. + let entries = &state.core.config.storage_entries; + let active = &state.core.active_backend_name; + if entries.iter().all(|e| e.name != dto.target_name) { + let available = if entries.is_empty() { + "(none)".to_string() + } else { + entries + .iter() + .map(|e| e.name.as_str()) + .collect::>() + .join(", ") + }; + return Err(AppError::bad_request(format!( + "unknown target entry `{}`. Available: [{available}]", + dto.target_name + ))); + } + if dto.target_name == *active { + return Err(AppError::bad_request(format!( + "target `{}` is the currently-active entry — pick a different entry to migrate to", + dto.target_name + ))); + } + trigger_storage_migration(state, Some(dto.target_name)).await } /// POST /api/admin/storage/migration/pause — pause a running migration. @@ -528,7 +555,11 @@ pub async fn pause_migration( pub async fn resume_migration( State(state): State>, ) -> Result { - trigger_storage_migration(state).await + // Resume path — no target_name in the body. The handler reads + // it from `params.target_name` stamped on the original Fresh + // open. Refuses gracefully via RunOutcome::Failed if there is + // no Paused row to resume. + trigger_storage_migration(state, None).await } /// POST /api/admin/storage/migration/verify — post-migration integrity check. @@ -590,6 +621,7 @@ pub async fn verify_migration( /// itself is fire-and-forget. async fn trigger_storage_migration( state: Arc, + target_name: Option, ) -> Result { use crate::infrastructure::scheduler::JobRunArgs; use crate::infrastructure::services::storage_migration_service::STORAGE_MIGRATION_JOB_NAME; @@ -597,14 +629,17 @@ async fn trigger_storage_migration( tracing::info!( target: "audit", event = "storage_migration.trigger_requested", + target_name = target_name.as_deref().unwrap_or(""), "👮🏻‍♂️ Admin triggered storage_migration" ); let registry = state.core.job_registry.clone(); + let args = JobRunArgs { + storage: target_name, + ..JobRunArgs::default() + }; tokio::spawn(async move { - registry - .trigger(STORAGE_MIGRATION_JOB_NAME, &JobRunArgs::default()) - .await; + registry.trigger(STORAGE_MIGRATION_JOB_NAME, &args).await; }); Ok(( @@ -2203,6 +2238,16 @@ pub struct TriggerJobQuery { 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 + /// `storage_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, } /// `POST /api/admin/jobs/{name}/trigger` — dispatch one run off-schedule. @@ -2249,6 +2294,7 @@ pub async fn trigger_job( let args = JobRunArgs { force: query.force, deep: query.deep, + storage: query.storage.clone(), }; // Jobs that can run for hours (storage_migration, future