feat(storage): jobs can choose storage to migrate/scan

This commit is contained in:
Edouard Vanbelle
2026-08-01 12:59:35 +02:00
parent 354e058114
commit 6b7bb67500
7 changed files with 385 additions and 49 deletions
+17
View File
@@ -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<usize>,
}
+28 -8
View File
@@ -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<dyn BlobStorageBackend> = 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<dyn BlobStorageBackend>,
/// 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
@@ -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<Option<String>, 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<String>,)> =
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#"
@@ -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<Option<String>, 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<Finding>,
progress_total: Option<u64>,
progress_kind: Option<ProgressKind>,
string_params: std::collections::HashMap<String, String>,
}
#[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<Option<String>, 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());
+12
View File
@@ -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<String>,
}
/// Uniform outcome the supervisor logs and stores for every job dispatch.
@@ -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<PgPool>,
/// 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<dyn BlobStorageBackend>,
storage_settings: Arc<StorageSettingsService>,
/// 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<NamedStorageEntry>,
/// 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<PgPool>,
source: Arc<dyn BlobStorageBackend>,
storage_settings: Arc<StorageSettingsService>,
active_backend_name: String,
storage_entries: Vec<NamedStorageEntry>,
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<Vec<u8>>,
) -> 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\": \"<entry>\"}`."
.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::<Vec<_>>()
.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:<missing-config>".to_string(),
},
StorageBackendType::Azure => match entry.azure.as_ref() {
Some(az) => format!(
"azure:{}/{}",
az.account_name.as_str(),
az.container.as_str()
),
None => "azure:<missing-config>".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'
+52 -6
View File
@@ -454,9 +454,36 @@ pub async fn get_migration_status(
)]
pub async fn start_migration(
State(state): State<Arc<AppState>>,
Json(_dto): Json<StartMigrationDto>,
Json(dto): Json<StartMigrationDto>,
) -> Result<impl IntoResponse, AppError> {
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::<Vec<_>>()
.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<Arc<AppState>>,
) -> Result<impl IntoResponse, AppError> {
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<AppState>,
target_name: Option<String>,
) -> Result<axum::response::Response, AppError> {
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("<resume>"),
"👮🏻‍♂️ 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<String>,
}
/// `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