From d7c19570a506a403f6cdc806dc4ed76722613a68 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sat, 1 Aug 2026 13:54:00 +0200 Subject: [PATCH] feat(storage): wire choice of storage --- frontend/src/lib/api/endpoints/admin.test.ts | 11 +- frontend/src/lib/api/endpoints/admin.ts | 75 +++-- .../src/routes/admin/[[tab]]/+page.svelte | 266 ++++++++++++------ src/application/dtos/settings_dto.rs | 65 ++++- .../services/storage_settings_service.rs | 68 ++++- src/common/di.rs | 14 +- .../services/backend_consistency_service.rs | 113 +++++++- .../services/blobs_consistency_service.rs | 114 +++++++- src/interfaces/api/handlers/admin_handler.rs | 131 +-------- src/interfaces/api/mod.rs | 3 +- 10 files changed, 574 insertions(+), 286 deletions(-) diff --git a/frontend/src/lib/api/endpoints/admin.test.ts b/frontend/src/lib/api/endpoints/admin.test.ts index 1df88227..5c4a6977 100644 --- a/frontend/src/lib/api/endpoints/admin.test.ts +++ b/frontend/src/lib/api/endpoints/admin.test.ts @@ -104,15 +104,8 @@ describe('admin test/probe endpoints', () => { await expect(admin.testStorage({ backend: 's3' })).resolves.toMatchObject({ connected: true }); }); - it('verifyMigration fills defaults and throws on error', async () => { - fetchMock.mockResolvedValue(okRes({ passed: true })); - await expect(admin.verifyMigration(10)).resolves.toMatchObject({ - passed: true, - sample_checked: 0 - }); - fetchMock.mockResolvedValue(errRes(500, {})); - await expect(admin.verifyMigration()).rejects.toThrow(/verify failed/); - }); + // verifyMigration retired in slice 7 of docs/plan/storage-multi-entry.md. + // Superseded by `POST /api/admin/jobs/blobs_consistency/trigger?storage=`. it('installPlugin posts a FormData bundle', async () => { fetchMock.mockResolvedValue(okRes({ id: 'com.example.hello' })); diff --git a/frontend/src/lib/api/endpoints/admin.ts b/frontend/src/lib/api/endpoints/admin.ts index 1f773c75..b15c5bae 100644 --- a/frontend/src/lib/api/endpoints/admin.ts +++ b/frontend/src/lib/api/endpoints/admin.ts @@ -466,6 +466,15 @@ export function saveOidc(body: Record): Promise { // ── Storage settings + migration ─────────────────────────────────────────── +export interface StorageEntrySummary { + name: string; + backend: string; + is_active: boolean; + encryption_enabled: boolean; + /** Human-readable physical hint (root_dir / bucket / container). */ + location_hint?: string | null; +} + export interface StorageSettings { backend: string; s3_endpoint_url?: string | null; @@ -479,6 +488,11 @@ export interface StorageSettings { total_blobs?: number; total_bytes_stored?: number; dedup_ratio?: number; + // Multi-entry view (slice 6 of docs/plan/storage-multi-entry.md). + // `entries` is empty for the legacy zero-entries path. + entries?: StorageEntrySummary[]; + active_entry_name?: string; + migration_readonly?: boolean; } export function getStorageSettings(): Promise { @@ -512,50 +526,33 @@ export function getMigration(): Promise { return apiJson('/api/admin/storage/migration', { credentials: 'same-origin' }); } -export function migrationAction(action: 'start' | 'pause' | 'resume'): Promise { +export function migrationAction( + action: 'start' | 'pause' | 'resume', + targetName?: string +): Promise { // `complete` was retired when the migration became a recoverable // job — Completed is the terminal `RunSummary.status`; there's - // nothing left to acknowledge. Post-migration cutover now happens - // via .env + restart, prompted by an inline hint on the admin - // storage tab (see `cutoverPending` in +page.svelte). - const body = action === 'start' ? { concurrency: 4 } : {}; + // nothing left to acknowledge. Post-migration cutover happens on + // operator restart (server re-boots on active_backend_name = the + // new entry; the boot-clear rule drops migration_readonly). + // + // `start` REQUIRES `targetName` in multi-entry mode — the backend + // rejects an unnamed start with 400 (see StartMigrationDto). + // `pause` and `resume` take no body (resume reads target_name + // from the paused run's params). + const body: Record = + action === 'start' ? { target_name: targetName ?? '', concurrency: 4 } : {}; return mutate(`/api/admin/storage/migration/${action}`, 'POST', body); } -/** Result of a `verify` integrity check (POST .../migration/verify). */ -export interface MigrationVerifyResult { - passed: boolean; - sample_checked: number; - pg_blob_count: number; - missing_in_target: string[]; - size_mismatches: string[]; -} - -/** - * Run an integrity verification pass over a sample of migrated blobs. Unlike - * the other migration actions this returns a structured result that the caller - * renders (passed / sample-checked / missing / size-mismatch counts). - */ -export async function verifyMigration(sampleSize = 100): Promise { - const res = await apiFetch('/api/admin/storage/migration/verify', { - method: 'POST', - credentials: 'same-origin', - headers: { ...JSON_HEADERS, ...getCsrfHeaders() }, - body: JSON.stringify({ sample_size: sampleSize }) - }); - if (!res.ok) { - const e = (await res.json().catch(() => ({}))) as { message?: string }; - throw new Error(e.message || `verify failed: ${res.status}`); - } - const r = (await res.json()) as Partial; - return { - passed: r.passed ?? false, - sample_checked: r.sample_checked ?? 0, - pg_blob_count: r.pg_blob_count ?? 0, - missing_in_target: r.missing_in_target ?? [], - size_mismatches: r.size_mismatches ?? [] - }; -} +// verifyMigration + MigrationVerifyResult retired in slice 7 of +// docs/plan/storage-multi-entry.md — the corresponding backend +// endpoint's sample-based check is superseded by +// `POST /api/admin/jobs/blobs_consistency/trigger?storage=`, +// which does a full walk against any named entry and integrates +// with the standard runs / findings admin surface. Trigger from +// the Jobs tab; the Storage tab drops the "Verify integrity" +// button. // ── Plugins ───────────────────────────────────────────────────────────── diff --git a/frontend/src/routes/admin/[[tab]]/+page.svelte b/frontend/src/routes/admin/[[tab]]/+page.svelte index e599fb1f..8b28a792 100644 --- a/frontend/src/routes/admin/[[tab]]/+page.svelte +++ b/frontend/src/routes/admin/[[tab]]/+page.svelte @@ -35,7 +35,6 @@ setUserRole, testOidc, testStorage, - verifyMigration, createExternalMount, deleteExternalMount, listExternalMounts, @@ -44,7 +43,6 @@ type AdminDashboard, type GeneratedKey, type MigrationStatus, - type MigrationVerifyResult, type OidcSettings, type OidcTestResult, type PluginInfo, @@ -519,15 +517,39 @@ stopMigrationPoll(); } } - async function doMigration(action: 'start' | 'pause' | 'resume') { + async function doMigration(action: 'start' | 'pause' | 'resume', targetName?: string) { try { - await migrationAction(action); + await migrationAction(action, targetName); await loadMigration(); } catch (e) { reportError(e); } } + // ── Multi-entry migration target picker (slice 6) ──────────────── + // + // Multi-entry mode requires the admin to name the target entry + // before starting a migration. Backend rejects an unnamed start + // with 400. Dropdown shows every non-active entry; picking one + // enables the Start button. + let migrationTarget = $state(''); + const availableTargets = $derived( + (storage?.entries ?? []).filter((e) => !e.is_active).map((e) => e.name) + ); + // Sync target when the entries list first appears — pick the first + // non-active entry by default so the operator can just click Start + // on a simple two-entry setup. + $effect(() => { + if (!migrationTarget && availableTargets.length > 0) { + migrationTarget = availableTargets[0]; + } + // Also unset when the previously-chosen target became active + // (cutover completed under our feet). + if (migrationTarget && !availableTargets.includes(migrationTarget)) { + migrationTarget = availableTargets[0] ?? ''; + } + }); + // ── Post-migration .env cutover hint ───────────────────────────── // // Migration copies blobs to the target backend, but boot-time @@ -595,22 +617,11 @@ } } - // Migration integrity verification (separate result panel). - let verifyResult = $state(null); - let verifyError = $state(null); - let verifying = $state(false); - async function doVerify() { - verifying = true; - verifyResult = null; - verifyError = null; - try { - verifyResult = await verifyMigration(100); - } catch (e) { - verifyError = errorMessage(e); - } finally { - verifying = false; - } - } + // Migration integrity verification retired in slice 7 — the + // sample-based /storage/migration/verify endpoint is replaced by + // `POST /api/admin/jobs/blobs_consistency/trigger?storage=`, + // a full walk. Operators trigger it from the Jobs tab. + const migrationPct = $derived( migration && migration.total_blobs > 0 ? Math.round((migration.migrated_blobs / migration.total_blobs) * 100) @@ -2147,6 +2158,62 @@

{t('admin.migration', 'Storage migration')}

+ + {#if storage?.entries && storage.entries.length > 0} + {#if storage.migration_readonly} +
+

+ + {t('admin.mig_readonly_title', 'Server in migration read-only mode')} +

+

+ {t( + 'admin.mig_readonly_body', + 'All writes (upload, rename, delete, share) are refused by AuthZ until the migration completes and you restart the server. Reads (browse, download) are unaffected.' + )} +

+
+ {/if} + + + + + + + + + + + + {#each storage.entries as entry (entry.name)} + + + + + + + + {/each} + +
{t('admin.entry_name', 'Entry')}{t('admin.entry_backend', 'Backend')}{t('admin.entry_location', 'Location')}{t('admin.entry_encryption', 'Encryption')}{t('admin.entry_status', 'Status')}
{entry.name}{entry.backend}{entry.location_hint ?? '—'} + {#if entry.encryption_enabled} + AES-256 + {:else} + — + {/if} + + {#if entry.is_active} + {t('admin.entry_active', 'active')} + {:else} + {t('admin.entry_inactive', 'available')} + {/if} +
+ {/if} {#if !migration}

{t('common.loading', 'Loading…')}

{:else} @@ -2179,13 +2246,46 @@ {/if}
- + {#if migration.status !== 'running' && migration.status !== 'paused' && migration.status !== 'completed'} - + {#if storage?.entries && storage.entries.length > 0} + + + {:else} + + {/if} {/if} {#if migration.status === 'running'} {/if} - - {#if migration.status === 'completed'} - - {/if} +
{#if cutoverPending} @@ -2265,55 +2353,6 @@
{/if} - - {#if verifyError} -
- {verifyError} -
- {:else if verifyResult} -
- - - {verifyResult.passed - ? t('admin.mig_verify_passed', 'Verification passed') - : t('admin.mig_verify_failed', 'Verification failed')} - - {#if verifyResult.passed} -

- {t( - 'admin.mig_verify_summary', - { checked: verifyResult.sample_checked, total: verifyResult.pg_blob_count }, - '{{checked}} blobs checked, {{total}} total in database' - )} -

- {:else} -

- {[ - verifyResult.missing_in_target.length - ? t( - 'admin.mig_verify_missing', - { n: verifyResult.missing_in_target.length }, - '{{n}} missing' - ) - : '', - verifyResult.size_mismatches.length - ? t( - 'admin.mig_verify_mismatch', - { n: verifyResult.size_mismatches.length }, - '{{n}} size mismatches' - ) - : '' - ] - .filter(Boolean) - .join(', ')} -

- {/if} -
- {/if} {/if} @@ -4226,6 +4265,49 @@ flex-wrap: wrap; } + .cutover-hint--readonly { + border-color: var(--color-danger-border, var(--color-border)); + background: var(--color-danger-bg, var(--color-bg-muted)); + } + + .entries-table { + width: 100%; + margin-bottom: var(--space-3); + border-collapse: collapse; + font-size: var(--text-sm, 0.875rem); + } + + .entries-table th, + .entries-table td { + padding: var(--space-2) var(--space-3); + border-bottom: 1px solid var(--color-border); + text-align: left; + } + + .entries-table th { + font-weight: 600; + color: var(--color-text-muted); + } + + .entries-table tr.entry-active { + background: var(--color-bg-muted); + } + + .migration-target-picker { + display: inline-flex; + align-items: center; + gap: var(--space-2); + margin-right: var(--space-2); + } + + .migration-target-picker select { + padding: var(--space-1) var(--space-2); + border: 1px solid var(--color-border); + border-radius: var(--radius-sm); + background: var(--color-bg); + color: var(--color-text); + } + .cutover-hint__note { flex: 1; min-width: 12rem; diff --git a/src/application/dtos/settings_dto.rs b/src/application/dtos/settings_dto.rs index fe04d4b2..ab90e1b0 100644 --- a/src/application/dtos/settings_dto.rs +++ b/src/application/dtos/settings_dto.rs @@ -143,10 +143,19 @@ pub struct DashboardStatsDto { // Storage Settings DTOs (Admin Panel) // ============================================================================ -/// Current storage settings returned to admin UI (secrets masked) +/// Current storage settings returned to admin UI (secrets masked). +/// +/// Post-multi-entry (`docs/plan/storage-multi-entry.md`) this carries +/// two shapes side-by-side: the legacy flat storage-config fields +/// (for the pre-multi-entry admin UI, until slice 6 completes the +/// form retirement), plus the new `entries` / `active_entry_name` / +/// `migration_readonly` view the multi-entry UI drives its entries-list, +/// migration-target-dropdown, and readonly-banner from. #[derive(Debug, Serialize, Deserialize)] pub struct StorageSettingsDto { - /// Active backend type: "local" or "s3" + /// Active backend type: "local" or "s3". Legacy flat-config field + /// — mirrors `entries[i where is_active].backend` for the active + /// entry when multi-entry is in use. pub backend: String, pub s3_endpoint_url: Option, pub s3_bucket: Option, @@ -163,6 +172,47 @@ pub struct StorageSettingsDto { pub total_blobs: u64, pub total_bytes_stored: u64, pub dedup_ratio: f64, + // ── Multi-entry view (slice 6) ── + /// All named storage entries declared in env. Empty when running + /// in legacy single-backend mode (`OXICLOUD_STORAGE_ENTRIES` + /// unset AND no legacy synthesis happened). Order matches + /// `_ENTRIES`. + pub entries: Vec, + /// Name of the entry the LIVE backend is currently bound to. + /// Populated as the boot-selected name (per + /// `CoreServices.active_backend_name`). Empty string for the + /// zero-entries legacy path (`"legacy"` sentinel). + pub active_entry_name: String, + /// Global read-only flag — when true, all write-adjacent + /// AuthZ checks refuse. Set by the migration handler at run + /// start; cleared by the boot-clear rule after operator + /// restart. Frontend renders a banner on the storage tab when + /// true. + pub migration_readonly: bool, +} + +/// Per-entry summary emitted in `StorageSettingsDto.entries`. Never +/// carries credentials — those live in env vars only. `is_active` +/// marks which entry the LIVE backend uses right now (matches +/// `active_entry_name` on the parent DTO). +#[derive(Debug, Serialize, Deserialize)] +pub struct StorageEntrySummaryDto { + pub name: String, + /// Backend type — "local" / "s3" / "azure". + pub backend: String, + /// True for exactly one entry (the entry the LIVE backend is on). + /// Frontend uses this to badge the active row and to exclude it + /// from the migration-target dropdown. + pub is_active: bool, + /// True when the entry has a per-entry encryption key. UI shows + /// a lock icon. Presence-only — the key bytes never leave the + /// server. + pub encryption_enabled: bool, + /// Human-readable physical location hint, if the backend surfaces + /// one (`root_dir` for Local, `bucket` for S3, `container` for + /// Azure). Cosmetic — helps the admin distinguish two Local + /// entries pointing at different disks. + pub location_hint: Option, } /// Request body for saving storage settings from the admin panel @@ -280,12 +330,11 @@ pub struct StartMigrationDto { pub concurrency: Option, } -/// Request body (empty) for `POST /api/admin/storage/migration/verify`. -#[derive(Debug, Serialize, Deserialize, ToSchema)] -pub struct VerifyMigrationDto { - /// Number of random blobs to sample-check (default: 100). - pub sample_size: Option, -} +// VerifyMigrationDto retired in slice 7 of +// docs/plan/storage-multi-entry.md — the corresponding endpoint's +// sample-based check is superseded by +// `blobs_consistency?storage=`, a full walk that emits +// structured findings per mismatch. // ============================================================================ // SMTP Settings DTOs (Admin Panel) diff --git a/src/application/services/storage_settings_service.rs b/src/application/services/storage_settings_service.rs index d33322cf..7d955e8c 100644 --- a/src/application/services/storage_settings_service.rs +++ b/src/application/services/storage_settings_service.rs @@ -2,11 +2,16 @@ use std::collections::HashMap; use std::sync::Arc; use uuid::Uuid; +use std::sync::atomic::{AtomicBool, Ordering}; + use crate::application::dtos::settings_dto::{ - SaveStorageSettingsDto, StorageSettingsDto, StorageTestResultDto, TestStorageConnectionDto, + SaveStorageSettingsDto, StorageEntrySummaryDto, StorageSettingsDto, StorageTestResultDto, + TestStorageConnectionDto, }; use crate::application::ports::blob_storage_ports::BlobStorageBackend; -use crate::common::config::{S3StorageConfig, StorageBackendType, StorageConfig}; +use crate::common::config::{ + NamedStorageEntry, S3StorageConfig, StorageBackendType, StorageConfig, +}; use crate::common::errors::{DomainError, ErrorKind}; use crate::domain::repositories::settings_repository::SettingsRepository; use crate::infrastructure::repositories::pg::SettingsPgRepository; @@ -22,18 +27,39 @@ pub struct StorageSettingsService { settings_repo: Arc, env_storage_config: StorageConfig, dedup_service: Arc, + /// Multi-entry snapshot from `AppConfig.storage_entries`. Populated + /// at DI time; immutable per-process (env can only change on + /// restart, per `docs/plan/storage-multi-entry.md`). Empty when + /// running in the pre-multi-entry legacy path. + storage_entries: Vec, + /// Name of the entry the LIVE backend is bound to (matches + /// `CoreServices.active_backend_name`). Empty string / "legacy" + /// for the zero-entries path. + active_entry_name: String, + /// Shared readonly flag — read into the admin DTO so the UI can + /// render a "server in migration read-only mode" banner. Same + /// atomic as `AppState.migration_readonly`; changes made by the + /// migration handler are visible without a DB round-trip. + migration_readonly: Arc, } impl StorageSettingsService { + #[allow(clippy::too_many_arguments)] pub fn new( settings_repo: Arc, env_storage_config: StorageConfig, dedup_service: Arc, + storage_entries: Vec, + active_entry_name: String, + migration_readonly: Arc, ) -> Self { Self { settings_repo, env_storage_config, dedup_service, + storage_entries, + active_entry_name, + migration_readonly, } } @@ -262,6 +288,27 @@ impl StorageSettingsService { crate::common::config::StorageBackendType::Azure => "azure", }; + // Project the multi-entry view. `is_active` is name-compared + // against the boot-selected `active_entry_name` (matches + // exactly one entry when we're in multi-entry mode; matches + // nothing when running the zero-entries legacy path, which + // is expected — the frontend hides the entries table then). + let entries: Vec = self + .storage_entries + .iter() + .map(|e| StorageEntrySummaryDto { + name: e.name.clone(), + backend: match e.backend { + StorageBackendType::Local => "local".to_string(), + StorageBackendType::S3 => "s3".to_string(), + StorageBackendType::Azure => "azure".to_string(), + }, + is_active: e.name == self.active_entry_name, + encryption_enabled: e.encryption_key_base64.is_some(), + location_hint: entry_location_hint(e), + }) + .collect(); + Ok(StorageSettingsDto { backend: backend_str.to_string(), s3_endpoint_url: effective.s3.as_ref().and_then(|s| s.endpoint_url.clone()), @@ -275,6 +322,9 @@ impl StorageSettingsService { total_blobs: stats.total_blobs, total_bytes_stored: stats.total_bytes_stored, dedup_ratio: stats.dedup_ratio, + entries, + active_entry_name: self.active_entry_name.clone(), + migration_readonly: self.migration_readonly.load(Ordering::Relaxed), }) } @@ -667,3 +717,17 @@ async fn run_backend_roundtrip( }, ) } + +/// Cosmetic human-readable identifier for an entry — the physical +/// location piece an admin uses to disambiguate two entries of the +/// same backend type. Never carries credentials. `None` when the +/// entry doesn't have a natural short label (S3 without a bucket, +/// which shouldn't happen because the parser rejects that shape at +/// boot). +fn entry_location_hint(entry: &NamedStorageEntry) -> Option { + match entry.backend { + StorageBackendType::Local => entry.root_dir.clone(), + StorageBackendType::S3 => entry.s3.as_ref().map(|s3| s3.bucket.clone()), + StorageBackendType::Azure => entry.azure.as_ref().map(|az| az.container.clone()), + } +} diff --git a/src/common/di.rs b/src/common/di.rs index 7903da7b..459537a2 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -1435,6 +1435,8 @@ impl AppServiceFactory { crate::infrastructure::services::blobs_consistency_service::BlobsConsistencyCheck::new( maintenance_pool.clone(), core.blob_backend.clone(), + core.config.storage_entries.clone(), + self.storage_path.clone(), ), ) .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) @@ -1453,6 +1455,8 @@ impl AppServiceFactory { crate::infrastructure::services::backend_consistency_service::BackendConsistencyCheck::new( maintenance_pool.clone(), core.blob_backend.clone(), + core.config.storage_entries.clone(), + self.storage_path.clone(), ), ) .register_recoverable_job(&core.job_registry, &job_store_provider_dyn) @@ -2191,11 +2195,19 @@ impl AppServiceFactory { app_state.admin_settings_service = Some(admin_svc.clone()); - // 9b-1b. Wire storage settings service (reuses same settings_repo) + // 9b-1b. Wire storage settings service (reuses same settings_repo). + // Multi-entry view fields (entries + active_entry_name + + // migration_readonly) are populated from the same sources + // the migration handler and AuthZ engine read from — one + // snapshot at DI, shared atomic for the readonly flag so + // changes are visible without a DB round-trip. let storage_settings_svc = Arc::new(StorageSettingsService::new( settings_repo.clone(), self.config.storage.clone(), app_state.core.dedup_service.clone(), + app_state.core.config.storage_entries.clone(), + app_state.core.active_backend_name.clone(), + app_state.migration_readonly.clone(), )); app_state.storage_settings_service = Some(storage_settings_svc.clone()); tracing::info!("Storage settings service initialized"); diff --git a/src/infrastructure/services/backend_consistency_service.rs b/src/infrastructure/services/backend_consistency_service.rs index 71d5d8a7..67f9b656 100644 --- a/src/infrastructure/services/backend_consistency_service.rs +++ b/src/infrastructure/services/backend_consistency_service.rs @@ -57,6 +57,12 @@ use crate::infrastructure::scheduler::{ pub const BACKEND_CONSISTENCY_JOB_NAME: &str = "backend_consistency"; +/// Same `params` JSONB key `blobs_consistency` uses — kept identical +/// so operators grepping run rows see the same convention across +/// both storage-audit tenants. +pub const PROBED_STORAGE_PARAM: &str = + crate::infrastructure::services::blobs_consistency_service::PROBED_STORAGE_PARAM; + /// 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 @@ -78,12 +84,32 @@ const _MAX_EXAMPLES: usize = 5; pub struct BackendConsistencyCheck { pool: Arc, + /// Default backend to enumerate when `args.storage` is `None` — + /// the live LIVE backend, injected at DI. `?storage=` + /// swaps in a fresh backend for the named entry (via + /// [`build_entry_backend`]). backend: Arc, + /// Snapshot of `AppConfig.storage_entries` for `?storage=` + /// resolution. Same rule blobs_consistency uses. + storage_entries: Vec, + /// `OXICLOUD_STORAGE_PATH` fallback for Local entries with no + /// `_ROOT_DIR`. Same fallback rule as boot. + storage_path_fallback: std::path::PathBuf, } impl BackendConsistencyCheck { - pub fn new(pool: Arc, backend: Arc) -> Self { - Self { pool, backend } + pub fn new( + pool: Arc, + backend: Arc, + storage_entries: Vec, + storage_path_fallback: std::path::PathBuf, + ) -> Self { + Self { + pool, + backend, + storage_entries, + storage_path_fallback, + } } pub async fn register_recoverable_job( @@ -138,9 +164,76 @@ impl RecoverableJobHandler for BackendConsistencyCheck { async fn run_resumable( &self, store: &dyn JobStore, - _args: &JobRunArgs, + args: &JobRunArgs, resume_cursor: Option>, ) -> RunOutcome { + // Resolve the backend to probe. Mirrors the shape + // `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}"), + }; + } + } + }; + let backend: Arc = match &probed_storage { + None => self.backend.clone(), + Some(name) => match self.storage_entries.iter().find(|e| &e.name == name) { + Some(entry) => crate::infrastructure::services::entry_backend::build_entry_backend( + entry, + &self.storage_path_fallback, + ), + 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!( + "storage entry `{name}` not found in OXICLOUD_STORAGE_ENTRIES. \ + Available: [{available}]" + ), + }; + } + }, + }; + if let Err(e) = backend.initialize().await { + return RunOutcome::Failed { + message: format!("probed backend init: {e}"), + }; + } + if let Some(name) = &probed_storage { + tracing::info!( + target: "audit", + event = "backend_consistency.probe_scoped", + run_id = %store.run_id(), + probed_storage = %name, + "backend_consistency enumerating entry `{name}` (via ?storage=) instead \ + of live backend" + ); + } + // Cursor = opaque backend continuation token, UTF-8-encoded. // Each backend defines its own format (local = shard/hash, // S3 = ListObjectsV2 continuation token, Azure = list @@ -190,11 +283,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { // splits canonical blobs (checked for orphan) from // "unknown" entries (sidecar files, foreign namespaces — // emitted as informational notices). - let page = match self - .backend - .list_blob_hashes(cursor.clone(), BATCH_SIZE) - .await - { + let page = match backend.list_blob_hashes(cursor.clone(), BATCH_SIZE).await { Ok(v) => v, Err(e) => { // Backend refuses / can't enumerate. First-batch @@ -219,7 +308,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { "anomaly", None, serde_json::json!({ - "backend": self.backend.backend_type(), + "backend": backend.backend_type(), "error": format!("{e}"), "note": "backend refused enumeration; no per-blob orphan probes attempted", }), @@ -229,7 +318,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { target: "oxicloud::consistency", event = "backend_consistency.unenumerable", run_id = %store.run_id(), - backend = self.backend.backend_type(), + backend = backend.backend_type(), "backend refused enumeration (typical during migration or on backends without list support)" ); return RunOutcome::Completed; @@ -265,7 +354,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { serde_json::json!({ "path": unknown.path, "mtime": unknown.mtime.map(|t| t.to_rfc3339()), - "backend": self.backend.backend_type(), + "backend": backend.backend_type(), "note": "non-canonical file in blob namespace (sidecar / wrong extension); not managed by dedup", }), ) @@ -331,7 +420,7 @@ impl RecoverableJobHandler for BackendConsistencyCheck { serde_json::json!({ "hash": entry.hash, "mtime": entry.mtime.map(|t| t.to_rfc3339()), - "backend": self.backend.backend_type(), + "backend": backend.backend_type(), }), ) .await; diff --git a/src/infrastructure/services/blobs_consistency_service.rs b/src/infrastructure/services/blobs_consistency_service.rs index f210e645..9112352b 100644 --- a/src/infrastructure/services/blobs_consistency_service.rs +++ b/src/infrastructure/services/blobs_consistency_service.rs @@ -47,6 +47,7 @@ //! pointing at reaped chunks) — already covered by //! `files_consistency::chunk_missing`. +use std::path::PathBuf; use std::sync::Arc; use async_trait::async_trait; @@ -54,13 +55,21 @@ use chrono::{DateTime, Duration, Utc}; use sqlx::PgPool; use crate::application::ports::blob_storage_ports::BlobStorageBackend; +use crate::common::config::NamedStorageEntry; 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 BLOBS_CONSISTENCY_JOB_NAME: &str = "blobs_consistency"; +/// `params` JSONB key under which the entry name being probed is +/// stashed on a Fresh run (matches `TARGET_NAME_PARAM` on +/// `storage_migration`). Resumed runs re-read it so a paused audit +/// survives restart without the admin re-specifying the target. +pub const PROBED_STORAGE_PARAM: &str = "probed_storage"; + /// Rows per batch. Blobs are numerous (millions on a busy install) /// but per-row work is one indexed backend probe + one indexed SQL /// ref-count query. 200 balances cancel-poll cadence against @@ -82,12 +91,35 @@ const AFFECTED_FILES_SAMPLE: i64 = 5; pub struct BlobsConsistencyCheck { pool: Arc, + /// The default backend to probe when `args.storage` is `None` — + /// the currently-active LIVE backend, injected at DI time. Runs + /// with `?storage=` build a fresh backend for the named + /// entry instead (via [`build_entry_backend`]). backend: Arc, + /// Snapshot of `AppConfig.storage_entries` used to resolve + /// `args.storage` to a `NamedStorageEntry`. Empty for the + /// legacy zero-entries path — `?storage=` runs then + /// fail-fast with a clear "no entries declared" message. + storage_entries: Vec, + /// Ambient `AppConfig.storage_path` — used as the `root_dir` + /// fallback for a Local target entry with no `_ROOT_DIR`. Same + /// fallback rule the boot path uses. + storage_path_fallback: PathBuf, } impl BlobsConsistencyCheck { - pub fn new(pool: Arc, backend: Arc) -> Self { - Self { pool, backend } + pub fn new( + pool: Arc, + backend: Arc, + storage_entries: Vec, + storage_path_fallback: PathBuf, + ) -> Self { + Self { + pool, + backend, + storage_entries, + storage_path_fallback, + } } pub async fn register_recoverable_job( @@ -147,6 +179,80 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { args: &JobRunArgs, resume_cursor: Option>, ) -> RunOutcome { + // Resolve the backend to probe. Two paths, mirroring the + // Fresh/Resumed split the storage_migration handler uses: + // + // * Fresh + args.storage=Some — probe that named entry + // instead of the live backend. Stamp probed_storage in + // params so a mid-audit restart resumes against the same + // entry without re-input. + // * Fresh + args.storage=None — probe the live backend + // (today's default; audit of what the app is actually + // using). + // * Resumed — read probed_storage from params; None means + // the original run was against the live backend. + 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}"), + }; + } + } + }; + let backend: Arc = match &probed_storage { + None => self.backend.clone(), + Some(name) => match self.storage_entries.iter().find(|e| &e.name == name) { + Some(entry) => build_entry_backend(entry, &self.storage_path_fallback), + 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!( + "storage entry `{name}` not found in OXICLOUD_STORAGE_ENTRIES. \ + Available: [{available}]" + ), + }; + } + }, + }; + if let Err(e) = backend.initialize().await { + return RunOutcome::Failed { + message: format!("probed backend init: {e}"), + }; + } + if let Some(name) = &probed_storage { + tracing::info!( + target: "audit", + event = "blobs_consistency.probe_scoped", + run_id = %store.run_id(), + probed_storage = %name, + "blobs_consistency probing entry `{name}` (via ?storage=) instead of \ + live backend" + ); + } + // 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. @@ -318,7 +424,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { // error (log + skip): a transient S3 network blip // shouldn't produce a flood of false data_loss // findings. - let exists = match self.backend.blob_exists(&row.hash).await { + let exists = match backend.blob_exists(&row.hash).await { Ok(v) => v, Err(e) => { tracing::warn!( @@ -369,7 +475,7 @@ impl RecoverableJobHandler for BlobsConsistencyCheck { // avoid mistaking it for "the hash we expect to // see on disk (i.e. what will fix this)". if args.deep { - match recompute_hash(self.backend.as_ref(), &row.hash).await { + match recompute_hash(backend.as_ref(), &row.hash).await { Ok(computed_hash) if computed_hash == row.hash => {} Ok(computed_hash) => { finding_count += 1; diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index a0925a97..2d1f91b1 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -19,7 +19,7 @@ use crate::application::dtos::settings_dto::{ AdminCreateUserDto, AdminResetPasswordDto, DashboardStatsDto, ListUsersQueryDto, MigrationStateDto, SaveOidcSettingsDto, SaveStorageSettingsDto, SendSmtpTestDto, SmtpInfoDto, SmtpTestResultDto, StartMigrationDto, TestOidcConnectionDto, TestStorageConnectionDto, - UpdateUserActiveDto, UpdateUserQuotaDto, UpdateUserRoleDto, VerifyMigrationDto, + UpdateUserActiveDto, UpdateUserQuotaDto, UpdateUserRoleDto, }; use crate::application::dtos::user_dto::{AdminUserSummaryDto, UserDto}; use crate::application::ports::authorization_ports::AuthorizationEngine; @@ -85,7 +85,9 @@ 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)) - .route("/storage/migration/verify", post(verify_migration)) + // 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=`. // Encryption key generation .route( "/settings/storage/generate-key", @@ -562,55 +564,15 @@ pub async fn resume_migration( trigger_storage_migration(state, None).await } -/// POST /api/admin/storage/migration/verify — post-migration integrity check. -/// -/// Independent of the copy job: samples `sample_size` random blobs -/// from `storage.blobs` and probes the currently-effective target -/// backend for their existence + declared size. Passes iff no -/// samples are missing and no sizes disagree. -#[utoipa::path( - post, - path = "/api/admin/storage/migration/verify", - responses( - (status = 200, description = "Verification result"), - (status = 401, description = "Unauthorized"), - (status = 403, description = "Admin required"), - (status = 500, description = "Verification failed") - ), - security(("bearerAuth" = [])), - tag = "admin" -)] -pub async fn verify_migration( - State(state): State>, - Json(dto): Json, -) -> Result { - let pool = state - .db_pool - .clone() - .ok_or_else(|| AppError::internal_error("Database not available"))?; - - let svc = state - .storage_settings_service - .as_ref() - .ok_or_else(|| AppError::internal_error("Storage settings service not available"))?; - - let target = svc - .build_effective_backend() - .await - .map_err(|e| AppError::internal_error(format!("Failed to build target backend: {}", e)))?; - target - .initialize() - .await - .map_err(|e| AppError::internal_error(format!("Target backend init failed: {}", e)))?; - - let sample_size = dto.sample_size.unwrap_or(100).clamp(1, 1000); - - let result = verify_backend_sample(target.as_ref(), pool.as_ref(), sample_size) - .await - .map_err(|e| AppError::internal_error(format!("Verification failed: {}", e)))?; - - Ok(Json(result)) -} +// verify_migration endpoint retired (slice 7 of +// docs/plan/storage-multi-entry.md). It was a sample-based sanity +// check against the currently-effective target backend; superseded +// by `POST /api/admin/jobs/blobs_consistency/trigger?storage=` +// which does a full walk against ANY named entry (not just the +// migration target), records structured findings per mismatch, and +// integrates with the standard runs / cancel / findings admin +// surface. Frontend "Verify integrity" button removed in the same +// slice. /// Shared body for `start` / `resume` — both funnel through /// `run_or_resume` via `JobRegistry::trigger`. Detaches into a @@ -652,73 +614,6 @@ async fn trigger_storage_migration( .into_response()) } -/// Verify a random sample of blobs against the given target backend. -/// Inlined from the retired `migration_job::verify_migration` — same -/// query, same result shape; the recoverable-run engine has no reason -/// to own an integrity check. -async fn verify_backend_sample( - target: &dyn crate::application::ports::blob_storage_ports::BlobStorageBackend, - pool: &sqlx::PgPool, - sample_size: usize, -) -> Result { - use crate::common::errors::DomainError; - - let pg_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs") - .fetch_one(pool) - .await - .unwrap_or(0); - - let sample_rows: Vec<(String, i64)> = - sqlx::query_as("SELECT hash, size FROM storage.blobs ORDER BY random() LIMIT $1") - .bind(sample_size as i64) - .fetch_all(pool) - .await - .map_err(|e| { - DomainError::internal_error("Migration", format!("Sample query failed: {}", e)) - })?; - - let mut missing = Vec::new(); - let mut size_mismatches = Vec::new(); - - for (hash, expected_size) in &sample_rows { - match target.blob_exists(hash).await { - Ok(false) => missing.push(hash.clone()), - Err(e) => { - tracing::warn!("blob_exists failed for {}: {}", hash, e); - missing.push(hash.clone()); - } - Ok(true) => { - if let Ok(actual_size) = target.blob_size(hash).await - && actual_size != *expected_size as u64 - { - size_mismatches.push(hash.clone()); - } - } - } - } - - let passed = missing.is_empty() && size_mismatches.is_empty(); - Ok(MigrationVerifyResult { - pg_blob_count: pg_count as u64, - sample_checked: sample_rows.len() as u64, - missing_in_target: missing, - size_mismatches, - passed, - }) -} - -/// Post-migration verification result — same shape as the retired -/// `migration_job::VerificationResult` (kept identical so the admin -/// UI's `MigrationVerifyResult` decoder needs no change). -#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)] -pub struct MigrationVerifyResult { - pub pg_blob_count: u64, - pub sample_checked: u64, - pub missing_in_target: Vec, - pub size_mismatches: Vec, - pub passed: bool, -} - /// Idle-state DTO — no run has been triggered yet. fn idle_migration_dto() -> MigrationStateDto { MigrationStateDto { diff --git a/src/interfaces/api/mod.rs b/src/interfaces/api/mod.rs index e9c510d7..25f8918b 100644 --- a/src/interfaces/api/mod.rs +++ b/src/interfaces/api/mod.rs @@ -225,7 +225,8 @@ use crate::interfaces::api::handlers::file_handler::MoveFilePayload; handlers::admin_handler::start_migration, handlers::admin_handler::pause_migration, handlers::admin_handler::resume_migration, - handlers::admin_handler::verify_migration, + // handlers::admin_handler::verify_migration retired in + // slice 7 — superseded by `blobs_consistency?storage=`. handlers::admin_handler::generate_encryption_key, // JobRegistry admin surface — production, always-on, // audit-logged. Retired the `/internal/trigger-*` handlers in