diff --git a/src/application/services/storage_settings_service.rs b/src/application/services/storage_settings_service.rs index 672df02b..6efac500 100644 --- a/src/application/services/storage_settings_service.rs +++ b/src/application/services/storage_settings_service.rs @@ -400,14 +400,23 @@ impl StorageSettingsService { entry, std::path::Path::new(&self.env_storage_config.root_dir), ); - let backend_kind = backend.backend_type().to_string(); + // Post-K2 (always-wrap): `backend.backend_type()` returns + // the outer wrapper's kind ("v1-plaintext" / "encrypted"), + // NOT the underlying storage backend. `health_check()` + // formats the wrapper-inner combo as + // `"()"` — that's the more informative + // string for the admin panel's Test-connection result. + // We use the wrapper-only string as a fallback for the + // health-check-failed branch, where there's no formatted + // status to draw from. + let fallback_backend_kind = backend.backend_type().to_string(); let status = match backend.health_check().await { Ok(s) => s, Err(e) => { return Ok(StorageTestResultDto { connected: false, message: format!("health-check failed: {e}"), - backend_type: backend_kind, + backend_type: fallback_backend_kind, available_bytes: None, roundtrip_passed: None, phase_reached: None, @@ -421,7 +430,7 @@ impl StorageSettingsService { let mut out = StorageTestResultDto { connected: status.connected, message: status.message, - backend_type: backend_kind, + backend_type: status.backend_type, available_bytes: status.available_bytes, roundtrip_passed: None, phase_reached: None, diff --git a/src/common/di.rs b/src/common/di.rs index 8a3e253e..6197fd94 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -2029,6 +2029,7 @@ impl AppServiceFactory { authorization: authorization.clone(), migration_readonly: migration_readonly.clone(), migration_progress: Arc::new(std::sync::RwLock::new(None)), + rotation_progress: Arc::new(std::sync::RwLock::new(None)), drive_repo: drive_repo.clone(), drive_management_service: Arc::new( crate::application::services::drive_management_service::DriveManagementService::new( @@ -2261,6 +2262,25 @@ impl AppServiceFactory { .register_recoverable_job(&app_state.core.job_registry, &job_store_provider_dyn) .await; + // K3: `storage_rotate` recoverable-job tenant. Same + // pattern as `storage_migration` but without the + // cutover/readonly plumbing — rotation writes in place on + // whichever entry the trigger endpoint names. Target name + // comes from `params.target_name` per run. + let _ = Arc::new( + crate::infrastructure::services::storage_rotate_service::StorageRotateService::new( + app_state + .maintenance_pool + .clone() + .expect("maintenance_pool set above"), + app_state.core.config.storage_entries.clone(), + self.storage_path.clone(), + app_state.rotation_progress.clone(), + ), + ) + .register_recoverable_job(&app_state.core.job_registry, &job_store_provider_dyn) + .await; + // 9b-2. Log whether system needs first-time admin setup if !admin_svc.is_system_initialized().await { tracing::warn!("╔══════════════════════════════════════════════════════════╗"); @@ -2795,6 +2815,15 @@ pub struct AppState { /// user's session banner about maintenance progress without /// polling. See `MigrationProgress` for the field shape. pub migration_progress: Arc>>, + /// Live progress snapshot for the storage-rotate handler + /// (`storage_rotate` — K3 of the storage-key-rotation plan). + /// `Some(_)` while a rotation is running; `None` otherwise. + /// Held separately from `migration_progress` so the + /// server-status header can broadcast the two states + /// independently: migration engages readonly mode, rotation + /// does not. Same `MigrationProgress` type — both are "walk + /// progress" fundamentally. + pub rotation_progress: Arc>>, /// Drive entity repository — `GET /api/drives`, the personal-drive /// lifecycle hook, and (post-D2) shared-drive creation flow all read /// through this. Backing table is `storage.drives`; membership is diff --git a/src/infrastructure/services/entry_backend.rs b/src/infrastructure/services/entry_backend.rs index 932947d6..0832e2d5 100644 --- a/src/infrastructure/services/entry_backend.rs +++ b/src/infrastructure/services/entry_backend.rs @@ -194,11 +194,47 @@ pub async fn resolve_active_entry<'a>( /// /// Both are boot-fatal and indicate a code (not config) bug, so /// panic is the honest response. -pub fn build_entry_backend( +/// Typed variant of [`build_entry_backend`] — returns the concrete +/// [`EncryptedBlobBackend`] wrapper so callers that need K3's +/// introspection API (`read_and_classify`, `head_format`, …) can hit +/// it directly without a downcast. +/// +/// Same construction path as `build_entry_backend`; the trait-object +/// version delegates through this. Preferred for job handlers +/// (`storage_rotate`) that need typed access. The trait-object +/// version stays for the DI hot-path where the caller only needs +/// the generic `BlobStorageBackend` contract. +pub fn build_entry_backend_typed( + entry: &NamedStorageEntry, + local_storage_path_fallback: &Path, +) -> Arc { + let base = build_base_backend(entry, local_storage_path_fallback); + let pairs = entry.encryption.clone().unwrap_or_default(); + let mode = match entry.head_cipher() { + Some(crate::common::config::CipherKind::AesGcm256) => "encrypted-v1", + _ => "plaintext-v1", + }; + tracing::info!( + "Storage entry `{}` — {} wrapper (pairs: {})", + entry.name, + mode, + pairs.len() + ); + Arc::new( + crate::infrastructure::services::encrypted_blob_backend::EncryptedBlobBackend::new( + base, pairs, + ), + ) +} + +/// Construct just the raw backend for the entry (no wrapper). Split +/// out of [`build_entry_backend`] so the typed variant can share the +/// switch on backend type without duplicating panic messages. +fn build_base_backend( entry: &NamedStorageEntry, local_storage_path_fallback: &Path, ) -> Arc { - let base: Arc = match entry.backend { + match entry.backend { StorageBackendType::Local => { let path = entry .root_dir @@ -227,27 +263,12 @@ pub fn build_entry_backend( }); Arc::new(crate::infrastructure::services::azure_blob_backend::AzureBlobBackend::new(az)) } - }; - - // v1 wrapper (Choice 1/B: always wrap). Every entry gets the - // header-aware read/write path — even entries with no - // `_ENCRYPTION_KEY` at all. This normalises the on-disk format - // going forward: all new writes carry the OXCPT v1 header, all - // reads magic-byte-dispatch (with legacy fallback for - // header-less pre-K2 blobs). Not backwards-compatible with - // pre-K2 code trying to read new writes — but Ed's called it: - // uniform format is worth the one-way door. - use crate::infrastructure::services::encrypted_blob_backend::EncryptedBlobBackend; - let pairs = entry.encryption.clone().unwrap_or_default(); - let mode = match entry.head_cipher() { - Some(crate::common::config::CipherKind::AesGcm256) => "encrypted-v1", - _ => "plaintext-v1", - }; - tracing::info!( - "Storage entry `{}` — {} wrapper (pairs: {})", - entry.name, - mode, - pairs.len() - ); - Arc::new(EncryptedBlobBackend::new(base, pairs)) + } +} + +pub fn build_entry_backend( + entry: &NamedStorageEntry, + local_storage_path_fallback: &Path, +) -> Arc { + build_entry_backend_typed(entry, local_storage_path_fallback) } diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index b7ab9895..cc6261bf 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -46,6 +46,7 @@ pub mod search_index; pub mod share_unlock_cookie; pub mod smtp_email_sender; pub mod storage_migration_service; +pub mod storage_rotate_service; pub mod swappable_blob_backend; pub mod thumbnail_service; #[cfg(test)] diff --git a/src/infrastructure/services/storage_rotate_service.rs b/src/infrastructure/services/storage_rotate_service.rs new file mode 100644 index 00000000..bf66ad7b --- /dev/null +++ b/src/infrastructure/services/storage_rotate_service.rs @@ -0,0 +1,489 @@ +//! Storage-format rotation as a recoverable-run tenant (K3 of +//! `docs/plan/storage-key-rotation.md`). +//! +//! Iterates `storage.blobs` for a target entry, decides per blob +//! whether the on-disk format matches what the entry's head pair +//! would write, and rewrites in place when it doesn't. Covers four +//! transitions with a single equality check: +//! +//! * Legacy blob (no `OXCPT` magic) → rewrite as v1 with the head +//! pair's format. +//! * v1 encrypted, decrypted under a pair-index other than head → +//! rewrite (key rotation). +//! * v1 plaintext with head=`aes:K` → rewrite (encrypt-in-place). +//! * v1 encrypted with head=`none:` → rewrite (decrypt-in-place). +//! +//! ### No readonly, no cutover +//! +//! `storage_rotate` is per-blob idempotent — repeat rewrites are +//! byte-safe (content-addressability holds; the wrapper always +//! produces the head format). Concurrent user writes coexist: they +//! land as head-format themselves, so when the walk reaches that +//! hash the classifier reports "already at head format" and the +//! decision tree collapses to `skip`. No app-wide read-only gate is +//! ever engaged — a critical improvement over `storage_migration`, +//! whose target-different-from-source cutover forces one. +//! +//! ### Restart survival +//! +//! Cursor + per-blob failure findings are persisted after every +//! batch. On restart, boot flips any abandoned `Running` row to +//! `Paused`; an admin trigger resumes from the checkpointed cursor. +//! The last checkpoint window (~100 blobs) re-processes; each of +//! those blobs is now head-format from the previous run's rewrite, +//! so the walk short-circuits without re-writing. Effectively free. +//! +//! ### Design notes +//! +//! * **Cursor** — UTF-8 hex of the last-processed blob hash (64 +//! chars). Same encoding as `storage_migration` and +//! `blobs_consistency`. +//! * **Target lookup** — the entry NAME is stashed in `params` at +//! Fresh-open time and re-read on Resume. The wrapper for that +//! entry is rebuilt at the top of every run via +//! `build_entry_backend_typed`; mid-run config changes are +//! ignored until the next run (mirrors `storage_migration`). +//! * **Per-blob failures don't fail the run** — each failure records +//! a `rotation_failed` finding (severity `data_loss` — the bytes +//! didn't get rewritten) and the walk continues. A run that +//! completes with zero findings is proof every blob is at head +//! format. +//! * **`?deep=true` is unused** — rotation has no slow variant. +//! Parameter accepted for uniformity with other tenants; ignored. + +use std::path::PathBuf; +use std::sync::Arc; + +use async_trait::async_trait; +use bytes::Bytes; +use sqlx::PgPool; + +use crate::application::ports::blob_storage_ports::BlobStorageBackend; +use crate::common::config::NamedStorageEntry; +use crate::common::migration_progress::MigrationProgress; +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, + RunStatus, record_or_log, +}; +use crate::infrastructure::services::entry_backend::build_entry_backend_typed; + +pub const STORAGE_ROTATE_JOB_NAME: &str = "storage_rotate"; + +/// The `params` JSONB key under which the run's target entry name is +/// stashed at Fresh-open time via `JobStore::set_string_param`. +/// Kept identical to `storage_migration`'s TARGET_NAME_PARAM so +/// operators grepping run rows see the same convention across both +/// storage-touching tenants. +pub const TARGET_NAME_PARAM: &str = "target_name"; + +/// Rows per batch. Matches `storage_migration` / `blobs_consistency` +/// so the checkpoint + cancel-poll cadence is uniform across tenants. +const BATCH_SIZE: i64 = 100; + +pub struct StorageRotateService { + pool: Arc, + /// Immutable per-deploy snapshot; used to look up the target + /// entry by name at run start. 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, + /// Shared in-memory progress snapshot for the server-status + /// header middleware. `Some(_)` while a rotation is + /// running/paused, `None` otherwise. Distinct from + /// `AppState.migration_progress` so the header can broadcast + /// migration + rotation states independently. + rotation_progress: Arc>>, +} + +impl StorageRotateService { + pub fn new( + pool: Arc, + storage_entries: Vec, + storage_path_fallback: PathBuf, + rotation_progress: Arc>>, + ) -> Self { + Self { + pool, + storage_entries, + storage_path_fallback, + rotation_progress, + } + } + + /// Chainable self-registration — mirrors the `*_consistency` + /// tenants and `storage_migration`. On-demand only (no periodic + /// tick). + pub async fn register_recoverable_job( + self: Arc, + registry: &JobRegistry, + provider: &Arc, + ) -> Arc { + registry + .register_recoverable_job(self.clone(), provider.clone(), None) + .await; + self + } +} + +#[async_trait] +impl RecoverableJobHandler for StorageRotateService { + fn name(&self) -> &str { + STORAGE_ROTATE_JOB_NAME + } + + /// Definitive count — one row per blob. Same query as + /// `storage_migration::count_total`; the two walk the same rows. + async fn count_total(&self) -> Option { + let row: Result<(i64,), sqlx::Error> = sqlx::query_as("SELECT COUNT(*) FROM storage.blobs") + .fetch_one(self.pool.as_ref()) + .await; + match row { + Ok((n,)) => Some(n.max(0) as u64), + Err(e) => { + tracing::debug!( + target: "oxicloud::rotate", + event = "storage_rotate.count_total_failed", + error = %e, + "count_total failed — run will not surface a progress bar" + ); + None + } + } + } + + async fn run_resumable( + &self, + store: &dyn JobStore, + args: &JobRunArgs, + resume_cursor: Option>, + ) -> RunOutcome { + // Resolve target entry name — same shape as `storage_migration`. + let is_fresh = resume_cursor.is_none(); + let target_name = if is_fresh { + let Some(name) = args.storage.clone() else { + return RunOutcome::Failed { + message: "storage_rotate requires `target_name` on a fresh run — trigger via \ + POST /api/admin/storage/entries/{name}/rotate" + .to_string(), + }; + }; + if let Err(e) = store.set_string_param(TARGET_NAME_PARAM, &name).await { + return RunOutcome::Failed { + 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 — cancel + trigger \ + fresh." + ), + }; + } + Err(e) => { + return RunOutcome::Failed { + message: format!("read {TARGET_NAME_PARAM} from params: {e}"), + }; + } + } + }; + + // Look up the target entry. + 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!( + "target entry `{target_name}` not declared in `OXICLOUD_STORAGE_ENTRIES` — \ + available: {available}." + ), + }; + } + }; + + // Build the wrapper for this entry — typed so we can call + // `read_and_classify` + `head_format` directly. + let wrapper = build_entry_backend_typed(target_entry, &self.storage_path_fallback); + if let Err(e) = wrapper.initialize().await { + return RunOutcome::Failed { + message: format!("target entry `{target_name}` failed to initialize: {e}"), + }; + } + let head_format = wrapper.head_format(); + + tracing::info!( + target: "audit", + event = "storage_rotate.run_started", + run_id = %store.run_id(), + target_name = %target_name, + head_format = ?head_format, + resuming = !is_fresh, + "storage_rotate started on `{target_name}` (head_format = {head_format:?})" + ); + + // Seed the progress snapshot. Total = count_total's estimate; + // if that failed we still surface the header without a + // denominator so the banner shows "rotation in progress" at + // minimum. + let total = self.count_total().await.unwrap_or(0); + { + let mut guard = self + .rotation_progress + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *guard = Some(MigrationProgress::new(target_name.clone(), total)); + } + + let mut cursor: Option = match resume_cursor { + None => None, + Some(bytes) if bytes.is_empty() => None, + Some(bytes) => match String::from_utf8(bytes) { + Ok(s) => Some(s), + Err(e) => { + self.clear_progress(); + return RunOutcome::Failed { + message: format!("invalid cursor: not valid UTF-8: {e}"), + }; + } + }, + }; + + let mut rewritten_count = 0u64; + let mut skipped_count = 0u64; + let mut failed_count = 0u64; + + loop { + // Cooperative cancel poll between batches. + match store.status().await { + Ok(RunStatus::CancelRequested) => { + self.clear_progress(); + tracing::info!( + target: "oxicloud::rotate", + event = "storage_rotate.cancelled", + run_id = %store.run_id(), + rewritten = rewritten_count, + skipped = skipped_count, + failed = failed_count, + "storage_rotate cancelled cooperatively, pausing" + ); + return RunOutcome::Paused { + cursor: cursor + .as_ref() + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(), + }; + } + Ok(_) => {} + Err(e) => { + self.clear_progress(); + return RunOutcome::Failed { + message: format!("status poll: {e}"), + }; + } + } + + // Fetch the next batch. Same keyset pagination shape as + // `storage_migration` — `hash > $1` on the PK, index-only. + let rows: Vec<(String,)> = match sqlx::query_as( + r#" + SELECT hash + FROM storage.blobs + WHERE ($1::text IS NULL OR hash > $1) + ORDER BY hash + LIMIT $2 + "#, + ) + .bind(cursor.as_deref()) + .bind(BATCH_SIZE) + .fetch_all(self.pool.as_ref()) + .await + { + Ok(r) => r, + Err(e) => { + self.clear_progress(); + return RunOutcome::Failed { + message: format!("batch fetch: {e}"), + }; + } + }; + + if rows.is_empty() { + return self + .finish_completed( + store, + &target_name, + rewritten_count, + skipped_count, + failed_count, + ) + .await; + } + + for (hash,) in &rows { + // Read + classify in one round-trip. Failure here is + // a real read failure (e.g. blob missing on disk), + // recorded as a finding. + let (plaintext, current_format) = match wrapper.read_and_classify(hash).await { + Ok(pair) => pair, + Err(e) => { + failed_count += 1; + tracing::warn!( + target: "oxicloud::rotate", + event = "storage_rotate.read_failed", + run_id = %store.run_id(), + hash = %hash, + error = %e, + "failed to read blob for classification; recording finding" + ); + record_or_log( + store, + STORAGE_ROTATE_JOB_NAME, + "rotation_failed", + "data_loss", + None, + serde_json::json!({ + "hash": hash, + "phase": "read", + "error": e.to_string(), + }), + ) + .await; + continue; + } + }; + + // The whole decision tree collapses to one equality + // check thanks to `BlobFormat`'s `PartialEq`. Six + // cases in the plan → one branch here. + if current_format == head_format { + skipped_count += 1; + continue; + } + + // Rewrite via the standard write path — atomic + // replace at the same object key. `put_blob_from_bytes` + // frames the plaintext with the head pair's format + // (encrypted-v1 or plaintext-v1) and hands the + // resulting bytes to the inner backend. + if let Err(e) = wrapper + .put_blob_from_bytes(hash, Bytes::from(plaintext.to_vec())) + .await + { + failed_count += 1; + tracing::warn!( + target: "oxicloud::rotate", + event = "storage_rotate.write_failed", + run_id = %store.run_id(), + hash = %hash, + error = %e, + "failed to rewrite blob; recording finding" + ); + record_or_log( + store, + STORAGE_ROTATE_JOB_NAME, + "rotation_failed", + "data_loss", + None, + serde_json::json!({ + "hash": hash, + "phase": "write", + "from": format!("{current_format:?}"), + "to": format!("{head_format:?}"), + "error": e.to_string(), + }), + ) + .await; + continue; + } + rewritten_count += 1; + } + + // Advance cursor + checkpoint. `delta_count` = work + // attempted this batch, so the progress bar advances even + // when a batch is dominated by skips (steady-state + // re-run) or failures. + let last_hash = rows.last().map(|(h,)| h.clone()).expect("non-empty rows"); + cursor = Some(last_hash.clone()); + let batch_len = rows.len() as u64; + if let Err(e) = store.checkpoint(last_hash.into_bytes(), batch_len).await { + self.clear_progress(); + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + { + let mut guard = self + .rotation_progress + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + if let Some(progress) = guard.as_mut() { + progress.bump(batch_len); + } + } + + if (rows.len() as i64) < BATCH_SIZE { + return self + .finish_completed( + store, + &target_name, + rewritten_count, + skipped_count, + failed_count, + ) + .await; + } + } + } +} + +impl StorageRotateService { + /// Terminal successful path — clear the header snapshot and log a + /// final audit line. Unlike `storage_migration::finish_completed` + /// there's no cutover / hot-swap step: rotation writes in place + /// on the entry that's already there. + async fn finish_completed( + &self, + store: &dyn JobStore, + target_name: &str, + rewritten: u64, + skipped: u64, + failed: u64, + ) -> RunOutcome { + self.clear_progress(); + tracing::info!( + target: "audit", + event = "storage_rotate.run_completed", + run_id = %store.run_id(), + target_name = %target_name, + rewritten = rewritten, + skipped = skipped, + failed = failed, + "storage_rotate completed on `{target_name}` — {rewritten} rewritten, {skipped} skipped, {failed} failed" + ); + RunOutcome::Completed + } + + fn clear_progress(&self) { + let mut guard = self + .rotation_progress + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *guard = None; + } +} diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index c000e520..433e3194 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -85,6 +85,14 @@ pub fn admin_routes() -> Router> { .route("/storage/migration/start", post(start_migration)) .route("/storage/migration/pause", post(pause_migration)) .route("/storage/migration/resume", post(resume_migration)) + // K3 (storage-key-rotation): per-entry rotate trigger. + // Normalises every blob on the named entry to its head-pair + // format (legacy → v1, plaintext ↔ encrypted, old-key → + // new-key). No readonly mode; safe under normal traffic. + .route( + "/storage/entries/{name}/rotate", + post(trigger_storage_rotate), + ) // NOTE: /storage/migration/verify retired in slice 7 (see the // comment near where `verify_migration` used to live). Use // `POST /api/admin/jobs/blobs_consistency/trigger?storage=`. @@ -621,6 +629,118 @@ async fn trigger_storage_migration( .into_response()) } +/// POST /api/admin/storage/entries/{name}/rotate — trigger the +/// `storage_rotate` recoverable job on a specific entry. +/// +/// Normalises every blob on `` to the entry's head-pair +/// format: legacy → v1, plaintext ↔ encrypted, old-key → new-key. +/// See `docs/plan/storage-key-rotation.md` §"The rotation job". +/// +/// Unlike migration, rotation does NOT engage read-only mode — +/// rewrites happen in place under normal traffic. Concurrent user +/// writes coexist safely. +/// +/// The handler validates the entry name synchronously (400 on +/// unknown entry); the actual walk detaches into a +/// `tokio::spawn` so the HTTP call returns immediately. +#[utoipa::path( + post, + path = "/api/admin/storage/entries/{name}/rotate", + responses( + (status = 202, description = "Rotation dispatched"), + (status = 400, description = "Unknown entry"), + (status = 401, description = "Unauthorized"), + (status = 403, description = "Admin required") + ), + params( + ("name" = String, Path, description = "Storage entry name to rotate") + ), + security(("bearerAuth" = [])), + tag = "admin" +)] +pub async fn trigger_storage_rotate( + State(state): State>, + axum::extract::Path(name): axum::extract::Path, +) -> Result { + use crate::infrastructure::scheduler::JobRunArgs; + use crate::infrastructure::services::storage_migration_service::STORAGE_MIGRATION_JOB_NAME; + use crate::infrastructure::services::storage_rotate_service::STORAGE_ROTATE_JOB_NAME; + + // Synchronous entry-existence check — a bad name would fail the + // run anyway, but returning 400 here spares the operator an + // audit-log round-trip. + let entries = &state.core.config.storage_entries; + if entries.iter().all(|e| e.name != 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 storage entry `{name}`. Available: [{available}]" + ))); + } + + // Concurrency guard per plan: at most one encryption-touching + // recoverable run at a time across the whole app. Rotation + // rewrites blobs in place; migration copies + swaps; running + // both simultaneously could interleave writes on the same + // hash. Cheap check — `list_runs` limit 1 with the status + // filter is an index scan. + let provider = state.core.job_store_provider.clone(); + for job_name in [STORAGE_ROTATE_JOB_NAME, STORAGE_MIGRATION_JOB_NAME] { + let in_flight = provider + .list_runs(job_name, 5) + .await + .map_err(AppError::from)? + .into_iter() + .any(|r| { + matches!( + r.status, + crate::infrastructure::scheduler::RunStatus::Running + | crate::infrastructure::scheduler::RunStatus::Paused + | crate::infrastructure::scheduler::RunStatus::CancelRequested + ) + }); + if in_flight { + return Err(AppError::bad_request(format!( + "cannot start storage_rotate on `{name}` — `{job_name}` is already Running / \ + Paused / CancelRequested. Wait for it to finish (or cancel via \ + `POST /api/admin/jobs/{job_name}/cancel`)." + ))); + } + } + + tracing::info!( + target: "audit", + event = "storage_rotate.trigger_requested", + target_name = %name, + "👮🏻‍♂️ Admin triggered storage_rotate on `{name}`" + ); + + let registry = state.core.job_registry.clone(); + let args = JobRunArgs { + storage: Some(name.clone()), + ..JobRunArgs::default() + }; + tokio::spawn(async move { + registry.trigger(STORAGE_ROTATE_JOB_NAME, &args).await; + }); + + Ok(( + StatusCode::ACCEPTED, + Json(serde_json::json!({ + "message": format!("Rotation dispatched on `{name}` — poll GET /api/admin/jobs/{STORAGE_ROTATE_JOB_NAME} for progress"), + "detached": true, + })), + ) + .into_response()) +} + /// Idle-state DTO — no run has been triggered yet. fn idle_migration_dto() -> MigrationStateDto { MigrationStateDto { diff --git a/src/interfaces/middleware/server_status.rs b/src/interfaces/middleware/server_status.rs index 2df6c57a..80926376 100644 --- a/src/interfaces/middleware/server_status.rs +++ b/src/interfaces/middleware/server_status.rs @@ -9,23 +9,22 @@ //! //! ## Cost model //! -//! On the *hot path* (no migration running — the ~100% case in normal -//! operation) this middleware does: +//! On the *hot path* (no migration AND no rotation running — the +//! ~100% case in normal operation) this middleware does: //! 1. one `AtomicBool::load(Relaxed)` — sub-nanosecond; -//! 2. an early return when `false`. +//! 2. one `RwLock::read` on `rotation_progress` — uncontended; +//! 3. an early return when both are inactive. //! -//! No allocation, no lock, no formatting. Adds no measurable latency -//! at any user count. +//! No allocation, no formatting on the hot path. The rotation-check +//! `RwLock::read` is cheap because writers only fire on batch +//! checkpoints (~every 100 blobs); worst-case contention is +//! sub-microsecond. //! -//! On the *cold path* (migration in progress) this middleware does: -//! 1. the atomic load above; -//! 2. one `RwLock::read` (uncontended — writers are the migration -//! handler, one per batch every ~100 blobs); -//! 3. one small `serde_json::to_string` call on a 4-field struct -//! (a few dozen bytes); -//! 4. one header insertion. +//! On the *cold path* (migration OR rotation in progress) the +//! payload builder pulls the progress snapshot(s), formats a small +//! JSON struct (~a few dozen bytes) and inserts the header. //! -//! Total per-request work in this branch: microseconds. +//! Total per-request work on cold path: microseconds. use axum::extract::Request; use axum::extract::State; @@ -53,11 +52,20 @@ pub const SERVER_STATUS_HEADER: &str = "x-server-status"; struct HeaderPayload { readonly: bool, #[serde(skip_serializing_if = "Option::is_none")] - migration: Option, + migration: Option, + /// K3: independent of `readonly` — rotation does NOT engage the + /// app-wide read-only flag, so the frontend needs a distinct + /// signal to know "rotation is running, show the rotation + /// banner instead of migration banner". + #[serde(skip_serializing_if = "Option::is_none")] + rotation: Option, } +/// Shared progress shape used by both `migration` and `rotation` +/// header fields — same struct name, same JSON field names. Frontend +/// treats them identically at the render layer. #[derive(serde::Serialize)] -struct MigrationHeader { +struct ProgressHeader { // `target` is owned here — the RwLock guard is released before // serialisation, so a borrowed slice wouldn't survive. Names // are small (`[a-z0-9_-]{1,32}`) so the copy is trivial. @@ -67,49 +75,76 @@ struct MigrationHeader { percent: u8, } +impl ProgressHeader { + fn from_snapshot(p: &crate::common::migration_progress::MigrationProgress) -> Self { + Self { + target: p.target_name.clone(), + migrated: p.migrated_blobs, + total: p.total_blobs, + percent: p.percent, + } + } +} + pub async fn server_status_middleware( State(state): State>, request: Request, next: Next, ) -> Response { - // Hot-path fast return. When no migration is running the flag is - // false and there's nothing to emit — a bare atomic load and out. let readonly = state.migration_readonly.load(Ordering::Relaxed); + + // Rotation snapshot check — cheap uncontended `read`; if `None` + // and readonly is also false, hot-path returns without a header. + let rotation_active = state + .rotation_progress + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .is_some(); + let mut response = next.run(request).await; - if !readonly { + if !readonly && !rotation_active { return response; } - // Cold path — build the payload from the shared progress - // snapshot. If the snapshot is absent (readonly is true but the - // handler hasn't seeded progress yet, or a restart-during- - // migration scenario) we still emit `readonly: true` so the - // banner shows — the frontend renders a "maintenance in progress" - // message even when specific numbers aren't available. + // Cold path — build the payload from whichever snapshots are + // active. `readonly:true` fires the migration banner even if + // the migration handler hasn't seeded its progress yet + // (restart-mid-migration scenario). `rotation` is populated + // independently. let payload = { - let guard = state - .migration_progress - .read() - .unwrap_or_else(std::sync::PoisonError::into_inner); + let migration = if readonly { + state + .migration_progress + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + .map(ProgressHeader::from_snapshot) + } else { + None + }; + let rotation = if rotation_active { + state + .rotation_progress + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .as_ref() + .map(ProgressHeader::from_snapshot) + } else { + None + }; HeaderPayload { - readonly: true, - migration: guard.as_ref().map(|p| MigrationHeader { - target: p.target_name.clone(), - migrated: p.migrated_blobs, - total: p.total_blobs, - percent: p.percent, - }), + readonly, + migration, + rotation, } }; - // `serde_json::to_string` on this 4-field struct is a few - // dozen-byte allocation — negligible against the response body. - // A serialize failure here would be a programming bug (all - // fields are trivially serializable), so we degrade to a - // minimal `readonly: true` string rather than skipping the - // header entirely. + // `serde_json::to_string` on this struct is a few dozen-byte + // allocation — negligible against the response body. A + // serialize failure here would be a programming bug, so we + // degrade to a minimal string rather than skipping the header. let value = - serde_json::to_string(&payload).unwrap_or_else(|_| r#"{"readonly":true}"#.to_string()); + serde_json::to_string(&payload).unwrap_or_else(|_| r#"{"readonly":false}"#.to_string()); if let Ok(header_value) = HeaderValue::from_str(&value) { response .headers_mut() diff --git a/tests/api/admin_jobs.hurl b/tests/api/admin_jobs.hurl index 35991bc3..78f7a850 100644 --- a/tests/api/admin_jobs.hurl +++ b/tests/api/admin_jobs.hurl @@ -96,15 +96,17 @@ jsonpath "$..interval_ms" count == 3 # wrapped by RecoverableAdapter so they appear here alongside the # periodics) + 1 coordinator (consistency_batch — a plain # JobHandler that dispatches every registered `*_consistency`) + -# 1 on-demand admin op (storage_migration — recoverable, no -# periodic tick, triggered by the admin panel's backend cutover). +# 2 on-demand admin ops (storage_migration — the readonly-mode + +# cutover backend swap; storage_rotate — K3, in-place per-blob +# format normalisation, no readonly). # Bump when a new tenant registers. -jsonpath "$..running" count == 11 +jsonpath "$..running" count == 12 jsonpath "$[*].name" contains "drives_consistency" jsonpath "$[*].name" contains "folders_consistency" jsonpath "$[*].name" contains "files_consistency" jsonpath "$[*].name" contains "consistency_batch" jsonpath "$[*].name" contains "storage_migration" +jsonpath "$[*].name" contains "storage_rotate" # ───────────────────────────────────────────────────────────── diff --git a/tests/api/storage_multi_entry.hurl b/tests/api/storage_multi_entry.hurl index d12c0d23..49bfe561 100644 --- a/tests/api/storage_multi_entry.hurl +++ b/tests/api/storage_multi_entry.hurl @@ -87,7 +87,13 @@ Content-Type: application/json HTTP 200 [Asserts] jsonpath "$.connected" == true -jsonpath "$.backend_type" == "local" +# Post-K2 (storage-key-rotation): every entry is wrapped in the v1 +# blob-format decorator, so `backend_type` reports the WRAPPER's kind +# in `"()"` form. `local_main` is unencrypted → wrapper +# is `v1-plaintext`. Match on the inner name via `contains` so the +# assertion survives future wrapper renames. +jsonpath "$.backend_type" contains "local" +jsonpath "$.backend_type" contains "v1-plaintext" jsonpath "$.roundtrip_passed" == true jsonpath "$.phase_reached" == "cleanup_ok"