diff --git a/Cargo.toml b/Cargo.toml index ce8f9b73..491fbd67 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -877,7 +877,21 @@ strip = true [profile.dev] opt-level = 1 -debug = true +# `line-tables-only` keeps file:line in panic backtraces (what you +# actually need on a long-running server) while dropping the rest of +# DWARF, which is worthless without a debugger. On this crate that +# takes target/debug/deps from ~58 GB to ~15-20 GB. Combined with +# `split-debuginfo = "unpacked"` (macOS-friendly: what little debug +# info remains lands in external .dSYM bundles that the linker +# doesn't embed in every .rlib), a full rebuild fits comfortably. +debug = "line-tables-only" +split-debuginfo = "unpacked" +# Incremental compilation caches per-function IR fingerprints so a +# small edit only recompiles what changed. On a single-crate rebuild +# (oxicloud is one crate) the savings are modest — worth < the ~7 GB +# incremental/ cache costs on disk. Rust-analyzer uses `cargo check`, +# which has its own cache, so LSP responsiveness is unaffected. +incremental = false [profile.bench] lto = "fat" diff --git a/frontend/src/lib/api/endpoints/admin.ts b/frontend/src/lib/api/endpoints/admin.ts index 8d9ca09b..1f773c75 100644 --- a/frontend/src/lib/api/endpoints/admin.ts +++ b/frontend/src/lib/api/endpoints/admin.ts @@ -384,13 +384,28 @@ export interface SmtpTestResult { error?: string; } -/** Result of POST .../settings/storage/test — the S3 connection probe. */ +/** + * Result of POST .../settings/storage/test. Combines reachability + * (`connected` — HEAD bucket / statfs) with a full read/write round- + * trip (`roundtrip_passed` — PUT + GET + verify + DELETE). Overall + * pass = both true. Round-trip fields are absent when the round-trip + * wasn't attempted (typically because reachability already failed). + * `phase_reached` names the last successful round-trip step: + * `initialize` | `put_ok` | `exists_ok` | `get_ok` | `verify_ok` | + * `cleanup_ok`. + */ export interface StorageTestResult { connected?: boolean; success?: boolean; backend_type?: string; available_bytes?: number | null; message?: string; + roundtrip_passed?: boolean; + phase_reached?: string; + bytes_written?: number; + bytes_read?: number; + roundtrip_elapsed_ms?: number; + cleanup_ok?: boolean; } export async function sendSmtpTest(to: string): Promise { @@ -497,7 +512,12 @@ export function getMigration(): Promise { return apiJson('/api/admin/storage/migration', { credentials: 'same-origin' }); } -export function migrationAction(action: 'start' | 'pause' | 'resume' | 'complete'): Promise { +export function migrationAction(action: 'start' | 'pause' | 'resume'): 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 } : {}; return mutate(`/api/admin/storage/migration/${action}`, 'POST', body); } diff --git a/frontend/src/routes/admin/[[tab]]/+page.svelte b/frontend/src/routes/admin/[[tab]]/+page.svelte index a8bb88ea..e599fb1f 100644 --- a/frontend/src/routes/admin/[[tab]]/+page.svelte +++ b/frontend/src/routes/admin/[[tab]]/+page.svelte @@ -466,18 +466,29 @@ storageMsg = null; try { const r: StorageTestResult = await testStorage(storageBody()); + // Backend now performs BOTH reachability (health-check) + // and a full read/write round-trip. `connected` gets + // flipped to false by the service if the round-trip + // itself fails, so a single boolean covers the whole + // pass/fail signal. `roundtrip_passed` distinguishes the + // two flavours of failure for the operator. const ok = r.connected ?? r.success ?? false; if (ok) { - let text = t('admin.storage_test_success', 'Connection successful'); + let text = t('admin.storage_test_success', 'Connection + read/write OK'); if (r.backend_type) text += ` (${r.backend_type})`; + if (r.roundtrip_elapsed_ms != null) text += ` — round-trip ${r.roundtrip_elapsed_ms} ms`; if (r.available_bytes != null) - text += ` — ${formatBytes(r.available_bytes)} ${t('admin.available', 'available')}`; + text += ` · ${formatBytes(r.available_bytes)} ${t('admin.available', 'available')}`; + if (r.cleanup_ok === false) + text += ` · ⚠ cleanup DELETE failed — orphan test blob left on backend`; storageMsg = { text, ok: true }; } else { - storageMsg = { - text: `${t('admin.storage_test_failure', 'Connection failed')}: ${r.message ?? ''}`, - ok: false - }; + const phase = r.phase_reached ? ` [phase: ${r.phase_reached}]` : ''; + const label = + r.roundtrip_passed === false + ? t('admin.storage_test_failure', 'Read/write test failed') + : t('admin.storage_test_failure', 'Connection failed'); + storageMsg = { text: `${label}${phase}: ${r.message ?? ''}`, ok: false }; } } catch (e) { storageMsg = { text: errorMessage(e), ok: false }; @@ -508,7 +519,7 @@ stopMigrationPoll(); } } - async function doMigration(action: 'start' | 'pause' | 'resume' | 'complete') { + async function doMigration(action: 'start' | 'pause' | 'resume') { try { await migrationAction(action); await loadMigration(); @@ -517,6 +528,73 @@ } } + // ── Post-migration .env cutover hint ───────────────────────────── + // + // Migration copies blobs to the target backend, but boot-time + // backend selection reads env vars only — never the DB config the + // admin filled in. So the app keeps running on the SOURCE backend + // even after the copy completes. To actually cut over, the + // operator has to add the equivalent env vars to `.env` and + // restart. This hint block spells out those lines with a + // copy-to-clipboard button. + // + // Shown only when: + // - a migration has completed successfully, AND + // - the live backend still differs from the configured target + // (so we're actually pending cutover), AND + // - the backend env var isn't ALREADY overriding (which would + // mean the admin already updated .env or the platform sets it). + const cutoverPending = $derived( + migration?.status === 'completed' && + !!storage && + storage.current_backend != null && + storage.current_backend !== storage.backend && + !(storage.env_overrides ?? []).includes('backend') + ); + + // Env-var lines the admin needs to paste. Credentials are NEVER + // echoed — the storage-settings DTO only returns `_set` booleans + // for access/secret keys (not the values), so we render a + // placeholder line the admin fills in from their own records. + // Local backend still gets a line for completeness, but a Local + // deployment typically has no reason to explicitly set the var + // (default is Local). + const cutoverEnvLines = $derived.by((): string[] => { + if (!storage) return []; + const lines: string[] = []; + switch (storage.backend) { + case 's3': + lines.push('OXICLOUD_STORAGE_BACKEND=s3'); + if (storage.s3_endpoint_url) + lines.push(`OXICLOUD_S3_ENDPOINT_URL=${storage.s3_endpoint_url}`); + if (storage.s3_bucket) lines.push(`OXICLOUD_S3_BUCKET=${storage.s3_bucket}`); + if (storage.s3_region) lines.push(`OXICLOUD_S3_REGION=${storage.s3_region}`); + if (storage.s3_access_key_set) + lines.push('OXICLOUD_S3_ACCESS_KEY='); + if (storage.s3_secret_key_set) + lines.push('OXICLOUD_S3_SECRET_KEY='); + if (storage.s3_force_path_style) lines.push('OXICLOUD_S3_FORCE_PATH_STYLE=true'); + break; + case 'local': + lines.push('OXICLOUD_STORAGE_BACKEND=local'); + break; + // Azure not yet exposed in the admin form; add here when it is. + } + return lines; + }); + + let cutoverCopied = $state(false); + async function copyCutoverEnv() { + try { + await navigator.clipboard.writeText(cutoverEnvLines.join('\n')); + cutoverCopied = true; + setTimeout(() => (cutoverCopied = false), 2000); + } catch { + // Clipboard permission denied — silent; the block is + // selectable so the operator can copy manually. + } + } + // Migration integrity verification (separate result panel). let verifyResult = $state(null); let verifyError = $state(null); @@ -2036,17 +2114,20 @@ data-testid="admin-storage-save-btn" disabled={storageBusy}>{t('common.save', 'Save')} - {#if sForm.backend === 's3'} - - {/if} + +
@@ -2120,7 +2201,10 @@ onclick={() => doMigration('resume')}>{t('admin.mig_resume', 'Resume')} {/if} - + {#if migration.status === 'completed'} {/if} + {#if cutoverPending} + +
+

+ + {t('admin.mig_cutover_title', 'Cutover pending — update .env and restart')} +

+

+ {t( + 'admin.mig_cutover_body', + { target: storage?.backend ?? '?', live: storage?.current_backend ?? '?' }, + 'Blobs are now on {{target}} but the server is still running on {{live}}. To switch, add these lines to your .env and restart the server.' + )} +

+ 
{cutoverEnvLines.join('\n')}
+
+ +

+ {t( + 'admin.mig_cutover_secret_note', + 'The access key and secret key are placeholders — paste the values you entered when saving these settings. Credentials are never displayed here.' + )} +

+
+
+ {/if} + {#if verifyError}
{verifyError} @@ -4070,6 +4195,43 @@ word-break: break-all; } + .cutover-hint { + margin-top: var(--space-3); + padding: var(--space-3); + border: 1px solid var(--color-warning-border, var(--color-border)); + border-radius: var(--radius-md); + background: var(--color-warning-bg, var(--color-bg-muted)); + } + + .cutover-hint h3 { + margin: 0 0 var(--space-2) 0; + font-size: var(--text-base, 1rem); + } + + .cutover-hint__lines { + margin: var(--space-2) 0; + padding: var(--space-2) var(--space-3); + background: var(--color-bg); + border: 1px solid var(--color-border); + border-radius: var(--radius-sm); + font-size: var(--text-xs, 0.75rem); + white-space: pre; + overflow-x: auto; + } + + .cutover-hint__actions { + display: flex; + align-items: flex-start; + gap: var(--space-3); + flex-wrap: wrap; + } + + .cutover-hint__note { + flex: 1; + min-width: 12rem; + margin: 0; + } + .smtp-test { display: flex; gap: var(--space-2); diff --git a/src/application/dtos/settings_dto.rs b/src/application/dtos/settings_dto.rs index bf0d6e36..05f35b6f 100644 --- a/src/application/dtos/settings_dto.rs +++ b/src/application/dtos/settings_dto.rs @@ -191,13 +191,50 @@ pub struct TestStorageConnectionDto { pub s3_force_path_style: Option, } -/// Result of a storage connection test +/// Result of a storage connection + round-trip test. +/// +/// `connected` is TRUE when the backend was reachable (health-check +/// passed — HEAD bucket / statfs). `roundtrip_passed` is TRUE when +/// the subsequent PUT → GET → verify → DELETE cycle succeeded — it +/// validates the exact permissions the migration job needs +/// (`s3:PutObject` + `s3:GetObject` + `s3:DeleteObject` on S3, disk +/// write permission on Local). All round-trip fields are `None` when +/// reachability failed (we don't attempt the round-trip if we can't +/// even HEAD the bucket). +/// +/// `phase_reached` names the last step that succeeded — on +/// `roundtrip_passed = false` it pinpoints where the failure hit +/// (`put_ok` → wrote but couldn't confirm; `exists_ok` → wrote + +/// confirmed but GET failed; etc.). `cleanup_ok = false` means the +/// backend was readable + writable but the test object may be +/// orphaned on it (~100 B, content-addressed — harmless, admin can +/// reap by hash). #[derive(Debug, Serialize, Deserialize)] pub struct StorageTestResultDto { pub connected: bool, pub message: String, pub backend_type: String, pub available_bytes: Option, + /// Set only when reachability passed AND a round-trip was + /// attempted. `Some(true)` = full write + read + verify success; + /// `Some(false)` = reachability OK, round-trip failed at + /// `phase_reached`; `None` = round-trip not attempted (typically + /// because reachability failed). + #[serde(skip_serializing_if = "Option::is_none")] + pub roundtrip_passed: Option, + /// Last round-trip phase completed successfully — one of + /// `initialize`, `put_ok`, `exists_ok`, `get_ok`, `verify_ok`, + /// `cleanup_ok`. `None` when round-trip wasn't attempted. + #[serde(skip_serializing_if = "Option::is_none")] + pub phase_reached: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub bytes_written: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub bytes_read: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub roundtrip_elapsed_ms: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub cleanup_ok: Option, } // ============================================================================ diff --git a/src/application/services/storage_settings_service.rs b/src/application/services/storage_settings_service.rs index 1d9fbf94..d33322cf 100644 --- a/src/application/services/storage_settings_service.rs +++ b/src/application/services/storage_settings_service.rs @@ -6,11 +6,13 @@ use crate::application::dtos::settings_dto::{ SaveStorageSettingsDto, StorageSettingsDto, StorageTestResultDto, TestStorageConnectionDto, }; use crate::application::ports::blob_storage_ports::BlobStorageBackend; -use crate::common::config::{S3StorageConfig, StorageConfig}; +use crate::common::config::{S3StorageConfig, StorageBackendType, StorageConfig}; use crate::common::errors::{DomainError, ErrorKind}; use crate::domain::repositories::settings_repository::SettingsRepository; use crate::infrastructure::repositories::pg::SettingsPgRepository; +use crate::infrastructure::services::azure_blob_backend::AzureBlobBackend; use crate::infrastructure::services::dedup_service::DedupService; +use crate::infrastructure::services::local_blob_backend::LocalBlobBackend; use crate::infrastructure::services::s3_blob_backend::S3BlobBackend; /// Storage settings service — manages storage backend configuration via the admin panel. @@ -92,6 +94,95 @@ impl StorageSettingsService { } } + /// Physical-storage identity string. Two configs that yield the + /// same `storage_identity` point at the same physical location + /// (same disk directory, same S3 bucket, same Azure container) — + /// used by [`Self::is_source_target_identical`] to detect a no-op + /// migration where source and target are the same backend. + /// + /// Credentials are deliberately excluded: two configs with + /// different access keys pointing at the same bucket ARE the same + /// storage; a migration between them would be a wasted walk. The + /// same principle applies to fields that don't influence which + /// bytes get read/written (chunk sizes, retention days, etc.). + fn storage_identity(config: &StorageConfig) -> String { + match config.backend { + StorageBackendType::Local => format!("local:{}", config.root_dir), + StorageBackendType::S3 => match config.s3.as_ref() { + Some(s3) => format!( + "s3:{}/{}:path_style={}", + s3.endpoint_url.as_deref().unwrap_or("aws"), + s3.bucket, + s3.force_path_style, + ), + None => "s3:".to_string(), + }, + StorageBackendType::Azure => match config.azure.as_ref() { + Some(az) => format!( + "azure:{}/{}", + az.account_name.as_str(), + az.container.as_str(), + ), + None => "azure:".to_string(), + }, + } + } + + /// True iff the *effective* storage config points at the same + /// physical location as the *boot* config — i.e. the migration + /// would be a no-op that walks every blob and skips them all. + /// + /// The migration handler calls this at run start and refuses with + /// `RunOutcome::Failed` if it's true — otherwise a misclick on an + /// S3 deployment would issue one `HEAD` per blob for zero useful + /// work (and real cost). "Legitimate" same-type migrations (e.g. + /// `local:/data` → `local:/newdisk`, or S3 bucket A → S3 bucket B) + /// return false and proceed normally. + pub async fn is_source_target_identical(&self) -> Result { + let effective = self.load_effective_storage_config().await?; + Ok(Self::storage_identity(&self.env_storage_config) == Self::storage_identity(&effective)) + } + + /// Build a `BlobStorageBackend` matching the current *effective* + /// storage config (DB + env-var overrides + defaults). + /// + /// Distinct from `dedup_service.backend()`, which is the LIVE + /// backend the app booted with — this method reflects what the + /// admin has configured *now* and typically resolves to a + /// different backend during a migration (source = live, target = + /// effective). The returned handle is a fresh instance; the caller + /// must `.initialize()` it before first use. + pub async fn build_effective_backend( + &self, + ) -> Result, DomainError> { + let effective = self.load_effective_storage_config().await?; + match effective.backend { + StorageBackendType::Local => Ok(Arc::new(LocalBlobBackend::new(std::path::Path::new( + &effective.root_dir, + )))), + StorageBackendType::S3 => { + let s3 = effective.s3.as_ref().ok_or_else(|| { + DomainError::new( + ErrorKind::InvalidInput, + "Storage", + "S3 backend selected but no S3 configuration is present", + ) + })?; + Ok(Arc::new(S3BlobBackend::new(s3))) + } + StorageBackendType::Azure => { + let az = effective.azure.as_ref().ok_or_else(|| { + DomainError::new( + ErrorKind::InvalidInput, + "Storage", + "Azure backend selected but no Azure configuration is present", + ) + })?; + Ok(Arc::new(AzureBlobBackend::new(az))) + } + } + } + /// Load effective storage config: DB settings + env var overrides + defaults. pub async fn load_effective_storage_config(&self) -> Result { let db: HashMap = self.settings_repo.get_by_category("storage").await?; @@ -245,21 +336,44 @@ impl StorageSettingsService { Ok(()) } - /// Test a storage connection by building a temporary backend and calling health_check(). + /// Test a storage backend: reachability (health-check) followed by + /// a full read/write round-trip (see [`run_backend_roundtrip`]). + /// + /// Two-phase so the operator gets clean diagnosis: if the health- + /// check fails they know it's an auth/endpoint/bucket problem + /// (never even wrote a byte). If it passes but the round-trip + /// fails, they know reachability is fine and the permissions are + /// the gap. Round-trip fields on the result are `None` when we + /// didn't attempt it (health-check failed early). pub async fn test_storage_connection( &self, dto: TestStorageConnectionDto, ) -> Result { match dto.backend.as_str() { "local" => { - // Test local backend health via the current dedup service backend - let status = self.dedup_service.backend().health_check().await?; - Ok(StorageTestResultDto { + // Local: no per-DTO override for the root_dir (the + // form has no such field), so we test the live + // backend the app is running on. health_check() reports + // available_bytes via statfs; the round-trip validates + // disk write + read + delete permissions. + let backend = self.dedup_service.backend().clone(); + let status = backend.health_check().await?; + let mut out = StorageTestResultDto { connected: status.connected, message: status.message, backend_type: "local".to_string(), available_bytes: status.available_bytes, - }) + roundtrip_passed: None, + phase_reached: None, + bytes_written: None, + bytes_read: None, + roundtrip_elapsed_ms: None, + cleanup_ok: None, + }; + if out.connected { + attach_roundtrip(&mut out, backend.as_ref()).await; + } + Ok(out) } "s3" => { let bucket = dto.s3_bucket.as_deref().unwrap_or_default(); @@ -269,11 +383,19 @@ impl StorageSettingsService { message: "S3 bucket name is required".to_string(), backend_type: "s3".to_string(), available_bytes: None, + roundtrip_passed: None, + phase_reached: None, + bytes_written: None, + bytes_read: None, + roundtrip_elapsed_ms: None, + cleanup_ok: None, }); } // Build a temporary S3 backend from the DTO values, - // falling back to existing DB/env config for missing fields. + // falling back to existing DB/env config for missing + // fields — lets the admin test values entered but not + // yet saved (matches the current UX). let effective = self.load_effective_storage_config().await.ok(); let existing_s3 = effective.as_ref().and_then(|c| c.s3.as_ref()); @@ -312,20 +434,36 @@ impl StorageSettingsService { }; let backend = S3BlobBackend::new(&config); - match backend.health_check().await { - Ok(status) => Ok(StorageTestResultDto { + let mut out = match backend.health_check().await { + Ok(status) => StorageTestResultDto { connected: status.connected, message: status.message, backend_type: "s3".to_string(), available_bytes: status.available_bytes, - }), - Err(e) => Ok(StorageTestResultDto { + roundtrip_passed: None, + phase_reached: None, + bytes_written: None, + bytes_read: None, + roundtrip_elapsed_ms: None, + cleanup_ok: None, + }, + Err(e) => StorageTestResultDto { connected: false, message: format!("Connection failed: {}", e), backend_type: "s3".to_string(), available_bytes: None, - }), + roundtrip_passed: None, + phase_reached: None, + bytes_written: None, + bytes_read: None, + roundtrip_elapsed_ms: None, + cleanup_ok: None, + }, + }; + if out.connected { + attach_roundtrip(&mut out, &backend).await; } + Ok(out) } other => Err(DomainError::new( ErrorKind::InvalidInput, @@ -335,3 +473,197 @@ impl StorageSettingsService { } } } + +/// Populate the round-trip fields of `out` by executing a full +/// PUT → EXISTS → GET → VERIFY → DELETE cycle against `backend`. Only +/// called when reachability (`out.connected`) already passed — +/// keeping "wasn't even reachable" and "reachable but round-trip +/// failed" as distinct diagnoses. +/// +/// On round-trip failure, `out.message` is REPLACED with the round- +/// trip diagnosis (the pre-round-trip message was just "connection +/// succeeded" — round-trip failure supersedes it). On round-trip +/// success, `out.message` is REPLACED with the success confirmation +/// so the admin sees the strong claim, not the weaker "reachable". +async fn attach_roundtrip(out: &mut StorageTestResultDto, backend: &dyn BlobStorageBackend) { + let (passed, phase, written, read, elapsed_ms, cleanup_ok, message) = + run_backend_roundtrip(backend).await; + out.roundtrip_passed = Some(passed); + out.phase_reached = Some(phase); + out.bytes_written = Some(written); + out.bytes_read = Some(read); + out.roundtrip_elapsed_ms = Some(elapsed_ms); + out.cleanup_ok = Some(cleanup_ok); + if !passed { + // Reachability was fine — the round-trip is the reason to + // fail this test overall. Flip `connected` to false so the + // UI shows the whole test as failed, and surface the + // round-trip diagnosis in the message. + out.connected = false; + } + out.message = message; +} + +/// Full read/write round-trip against the given backend. PUT a tiny +/// unique object, verify existence, GET it back, check BLAKE3 +/// matches, DELETE it. Validates the exact permissions the migration +/// job needs (`s3:PutObject` + `s3:GetObject` + `s3:DeleteObject` on +/// S3, disk write on Local) — a stronger check than `health_check`. +/// +/// Returns the round-trip fields for [`StorageTestResultDto`]. Errors +/// are folded into the return value (not `Err`) so callers can +/// surface `phase_reached` diagnostics inline. +async fn run_backend_roundtrip( + backend: &dyn BlobStorageBackend, +) -> ( + /* passed */ bool, + /* phase_reached */ String, + /* bytes_written */ u64, + /* bytes_read */ u64, + /* elapsed_ms */ u64, + /* cleanup_ok */ bool, + /* message */ String, +) { + use bytes::Bytes; + use futures::StreamExt; + use std::time::Instant; + + let started = Instant::now(); + + // Content-addressable — hash MUST be BLAKE3 of the payload. UUID + // + timestamp guarantees a fresh key on every test so we never + // collide with a real blob or stale test remnant. + let payload = format!( + "oxicloud-roundtrip-test-{}-{}", + uuid::Uuid::new_v4(), + chrono::Utc::now().timestamp_nanos_opt().unwrap_or(0), + ); + let payload_bytes = payload.into_bytes(); + let hash = blake3::hash(&payload_bytes).to_hex().to_string(); + let bytes_written = payload_bytes.len() as u64; + + if let Err(e) = backend + .put_blob_from_bytes(&hash, Bytes::from(payload_bytes.clone())) + .await + { + return ( + false, + "initialize".to_string(), + 0, + 0, + started.elapsed().as_millis() as u64, + false, + format!("put: {e}"), + ); + } + + match backend.blob_exists(&hash).await { + Ok(true) => {} + Ok(false) => { + let cleanup_ok = backend.delete_blob(&hash).await.is_ok(); + return ( + false, + "put_ok".to_string(), + bytes_written, + 0, + started.elapsed().as_millis() as u64, + cleanup_ok, + "PUT reported success but blob_exists returned false".to_string(), + ); + } + Err(e) => { + let cleanup_ok = backend.delete_blob(&hash).await.is_ok(); + return ( + false, + "put_ok".to_string(), + bytes_written, + 0, + started.elapsed().as_millis() as u64, + cleanup_ok, + format!("exists: {e}"), + ); + } + } + + let mut got: Vec = Vec::with_capacity(payload_bytes.len()); + match backend.get_blob_stream(&hash).await { + Ok(stream) => { + let mut stream = std::pin::pin!(stream); + while let Some(chunk) = stream.next().await { + match chunk { + Ok(bytes) => got.extend_from_slice(&bytes), + Err(e) => { + let cleanup_ok = backend.delete_blob(&hash).await.is_ok(); + return ( + false, + "exists_ok".to_string(), + bytes_written, + got.len() as u64, + started.elapsed().as_millis() as u64, + cleanup_ok, + format!("get stream: {e}"), + ); + } + } + } + } + Err(e) => { + let cleanup_ok = backend.delete_blob(&hash).await.is_ok(); + return ( + false, + "exists_ok".to_string(), + bytes_written, + 0, + started.elapsed().as_millis() as u64, + cleanup_ok, + format!("get: {e}"), + ); + } + } + let bytes_read = got.len() as u64; + + // VERIFY — recompute BLAKE3 on the received bytes. A hash + // mismatch means the backend returned different bytes than it + // stored (silent corruption in the round-trip). Extremely rare + // but the whole point of doing a byte-level test. + let got_hash = blake3::hash(&got).to_hex().to_string(); + if got_hash != hash { + let cleanup_ok = backend.delete_blob(&hash).await.is_ok(); + return ( + false, + "get_ok".to_string(), + bytes_written, + bytes_read, + started.elapsed().as_millis() as u64, + cleanup_ok, + format!( + "byte mismatch: wrote {bytes_written} bytes hash {hash}, read back {bytes_read} bytes hash {got_hash}" + ), + ); + } + + // CLEANUP — failure here does NOT flip `passed`. Read/write + // validation succeeded; the backend just left an orphan test + // blob (harmless — content-addressed, ~100 B). + let cleanup_ok = backend.delete_blob(&hash).await.is_ok(); + ( + true, + if cleanup_ok { + "cleanup_ok" + } else { + "verify_ok" + } + .to_string(), + bytes_written, + bytes_read, + started.elapsed().as_millis() as u64, + cleanup_ok, + if cleanup_ok { + "Round-trip OK: write + read + delete all succeeded".to_string() + } else { + format!( + "Round-trip OK (write + read validated), but cleanup DELETE failed — one orphan test blob left at hash {hash}" + ) + }, + ) +} diff --git a/src/common/di.rs b/src/common/di.rs index be862c6f..79412ccc 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -13,7 +13,6 @@ use crate::infrastructure::db::DbPools; use crate::application::services::admin_settings_service::AdminSettingsService; use crate::application::services::auth_application_service::AuthApplicationService; use crate::application::services::storage_settings_service::StorageSettingsService; -use crate::infrastructure::services::migration_blob_backend::MigrationState; use crate::application::ports::file_ports::FileUseCaseFactory; use crate::application::services::favorites_service::FavoritesService; @@ -1862,7 +1861,6 @@ impl AppServiceFactory { admin_settings_service: None, storage_settings_service: None, plugin_management, - migration_state: Arc::new(tokio::sync::RwLock::new(MigrationState::default())), trash_service, share_service, share_browse_service, @@ -2082,9 +2080,33 @@ impl AppServiceFactory { self.config.storage.clone(), app_state.core.dedup_service.clone(), )); - app_state.storage_settings_service = Some(storage_settings_svc); + app_state.storage_settings_service = Some(storage_settings_svc.clone()); 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). + let job_store_provider_dyn: Arc< + dyn crate::infrastructure::scheduler::JobStoreProvider, + > = app_state.core.job_store_provider.clone(); + let _ = Arc::new( + crate::infrastructure::services::storage_migration_service::StorageMigrationService::new( + app_state + .maintenance_pool + .clone() + .expect("maintenance_pool set above"), + app_state.core.blob_backend.clone(), + storage_settings_svc, + ), + ) + .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!("╔══════════════════════════════════════════════════════════╗"); @@ -2419,7 +2441,6 @@ pub struct AppState { pub plugin_management: Option>, pub storage_settings_service: Option>, - pub migration_state: Arc>, pub trash_service: Option>, pub share_service: Option>, pub share_browse_service: Option>, diff --git a/src/infrastructure/services/migration_blob_backend.rs b/src/infrastructure/services/migration_blob_backend.rs deleted file mode 100644 index dbe32231..00000000 --- a/src/infrastructure/services/migration_blob_backend.rs +++ /dev/null @@ -1,273 +0,0 @@ -//! `MigrationBlobBackend` — decorator that enables zero-downtime migration -//! between blob storage backends. -//! -//! During a migration the decorator writes to the **target** backend and reads -//! from **target-first-then-source** (dual-read). A background job -//! (see `migration_job.rs`) copies remaining blobs in the background. - -use std::future::Future; -use std::path::{Path, PathBuf}; -use std::pin::Pin; -use std::sync::Arc; - -use bytes::Bytes; -use chrono::{DateTime, Utc}; -use serde::Serialize; -use tokio::sync::RwLock; - -use crate::application::ports::blob_storage_ports::{ - BlobStorageBackend, BlobStream, StorageHealthStatus, -}; -use crate::common::errors::DomainError; - -// ── Migration state ──────────────────────────────────────────────── - -/// Progress of an ongoing (or completed) backend migration. -#[derive(Debug, Clone, Serialize)] -pub struct MigrationState { - pub status: MigrationStatus, - pub total_blobs: u64, - pub migrated_blobs: u64, - pub migrated_bytes: u64, - pub failed_blobs: Vec, - pub started_at: Option>, - pub completed_at: Option>, -} - -impl Default for MigrationState { - fn default() -> Self { - Self { - status: MigrationStatus::Idle, - total_blobs: 0, - migrated_blobs: 0, - migrated_bytes: 0, - failed_blobs: Vec::new(), - started_at: None, - completed_at: None, - } - } -} - -/// Status of the migration job. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] -#[serde(rename_all = "snake_case")] -pub enum MigrationStatus { - Idle, - Running, - Paused, - Completed, - Failed, -} - -// ── MigrationBlobBackend ─────────────────────────────────────────── - -/// A `BlobStorageBackend` decorator that proxies requests to a *source* -/// (old) and *target* (new) backend, enabling live migration. -pub struct MigrationBlobBackend { - source: Arc, - target: Arc, - state: Arc>, -} - -impl MigrationBlobBackend { - pub fn new( - source: Arc, - target: Arc, - state: Arc>, - ) -> Self { - Self { - source, - target, - state, - } - } - - pub fn state(&self) -> &Arc> { - &self.state - } - - pub fn source(&self) -> &Arc { - &self.source - } - - pub fn target(&self) -> &Arc { - &self.target - } -} - -/// Boxed future alias (same as in the trait module). -type BoxFut<'a, T> = Pin + Send + 'a>>; - -impl BlobStorageBackend for MigrationBlobBackend { - fn initialize(&self) -> BoxFut<'_, Result<(), DomainError>> { - Box::pin(async move { - self.target.initialize().await?; - // Source is already initialised; call anyway for idempotency. - self.source.initialize().await?; - Ok(()) - }) - } - - /// Writes go to **target** only. - fn put_blob(&self, hash: &str, source_path: &Path) -> BoxFut<'_, Result> { - let hash = hash.to_string(); - let path = source_path.to_path_buf(); - Box::pin(async move { self.target.put_blob(&hash, &path).await }) - } - - /// Writes bytes to **target** only. - fn put_blob_from_bytes(&self, hash: &str, data: Bytes) -> BoxFut<'_, Result> { - let hash = hash.to_string(); - Box::pin(async move { self.target.put_blob_from_bytes(&hash, data).await }) - } - - /// Unsynced writes go to **target** only (same as the synced variant). - fn put_blob_from_bytes_unsynced( - &self, - hash: &str, - data: Bytes, - ) -> BoxFut<'_, Result> { - let hash = hash.to_string(); - Box::pin(async move { self.target.put_blob_from_bytes_unsynced(&hash, data).await }) - } - - /// Durability sweep goes to **target**, where unsynced writes land. - fn sync_blobs(&self, hashes: &[String]) -> BoxFut<'_, Result<(), DomainError>> { - self.target.sync_blobs(hashes) - } - - /// Read from target first; fall back to source. - fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result> { - let hash = hash.to_string(); - Box::pin(async move { - match self.target.get_blob_stream(&hash).await { - Ok(stream) => Ok(stream), - Err(_) => self.source.get_blob_stream(&hash).await, - } - }) - } - - fn get_blob_range_stream( - &self, - hash: &str, - start: u64, - end: Option, - ) -> BoxFut<'_, Result> { - let hash = hash.to_string(); - Box::pin(async move { - match self.target.get_blob_range_stream(&hash, start, end).await { - Ok(stream) => Ok(stream), - Err(_) => self.source.get_blob_range_stream(&hash, start, end).await, - } - }) - } - - /// Delete from **both** backends (best-effort on source). - fn delete_blob(&self, hash: &str) -> BoxFut<'_, Result<(), DomainError>> { - let hash = hash.to_string(); - Box::pin(async move { - self.target.delete_blob(&hash).await?; - // Best-effort on source — ignore errors (blob may already be gone). - let _ = self.source.delete_blob(&hash).await; - Ok(()) - }) - } - - /// Exists in either backend. - fn blob_exists(&self, hash: &str) -> BoxFut<'_, Result> { - let hash = hash.to_string(); - Box::pin(async move { - if self.target.blob_exists(&hash).await? { - return Ok(true); - } - self.source.blob_exists(&hash).await - }) - } - - fn blob_size(&self, hash: &str) -> BoxFut<'_, Result> { - let hash = hash.to_string(); - Box::pin(async move { - match self.target.blob_size(&hash).await { - Ok(sz) => Ok(sz), - Err(_) => self.source.blob_size(&hash).await, - } - }) - } - - fn health_check(&self) -> BoxFut<'_, Result> { - Box::pin(async move { - let target_health = self.target.health_check().await?; - let source_health = self.source.health_check().await?; - Ok(StorageHealthStatus { - connected: target_health.connected && source_health.connected, - backend_type: format!( - "migration({} → {})", - source_health.backend_type, target_health.backend_type - ), - message: format!( - "Source: {} | Target: {}", - source_health.message, target_health.message - ), - available_bytes: target_health.available_bytes, - }) - }) - } - - fn backend_type(&self) -> &'static str { - "migration" - } - - /// Reads are served target-first (see `get_blob_stream`), so adopt the - /// target's read-ahead. - fn read_prefetch(&self) -> usize { - self.target.read_prefetch() - } - - fn local_blob_path(&self, hash: &str) -> Option { - // Prefer target, fall back to source. - self.target - .local_blob_path(hash) - .or_else(|| self.source.local_blob_path(hash)) - } - - /// Enumeration during migration is intentionally REFUSED. Both - /// source and target legitimately hold bytes concurrently - /// mid-migration: a blob copied to target but not yet deleted - /// from source would be reported "twice"; a blob in-flight from - /// source to target could be flagged as orphan on whichever - /// side the consistency scan doesn't walk. There's no single - /// authoritative "what's on the backend" answer while a - /// migration is running. - /// - /// Operators wanting to run `backend_consistency` during a - /// migration should either wait for the migration to complete - /// (target becomes authoritative) or cancel it. The - /// `operation_not_supported` error is surfaced by the tenant as - /// a single run-level `backend_unenumerable` finding — no - /// per-blob probes attempted. - fn list_blob_hashes( - &self, - _cursor: Option, - _limit: usize, - ) -> Pin< - Box< - dyn std::future::Future< - Output = Result< - crate::application::ports::blob_storage_ports::BlobListPage, - DomainError, - >, - > + Send - + '_, - >, - > { - Box::pin(async { - Err(DomainError::operation_not_supported( - "list_blob_hashes", - "backend_consistency cannot enumerate while a storage \ - migration is in progress — source and target hold bytes \ - concurrently; wait for migration completion or cancel it \ - before running the scan", - )) - }) - } -} diff --git a/src/infrastructure/services/migration_job.rs b/src/infrastructure/services/migration_job.rs deleted file mode 100644 index c2cd3673..00000000 --- a/src/infrastructure/services/migration_job.rs +++ /dev/null @@ -1,240 +0,0 @@ -//! Background migration job — copies blobs from a source backend to a target -//! backend with configurable concurrency and progress tracking. - -use std::sync::Arc; - -use futures::StreamExt; -use serde::Serialize; -use sqlx::PgPool; -use tokio::sync::RwLock; - -use crate::application::ports::blob_storage_ports::BlobStorageBackend; -use crate::common::errors::DomainError; -use crate::infrastructure::services::migration_blob_backend::{MigrationState, MigrationStatus}; - -/// Run the migration: stream all blob hashes from `storage.blobs` and copy -/// each one from `source` to `target`. -/// -/// * The job respects `Paused` / `Failed` status in `state` — it will stop -/// streaming when the status is no longer `Running`. -/// * Errors on individual blobs are logged and collected in `failed_blobs` -/// but do **not** abort the full run. -/// * `concurrency` controls `buffer_unordered` parallelism (default: 4). -pub async fn run_migration( - source: Arc, - target: Arc, - pool: Arc, - state: Arc>, - concurrency: usize, -) -> Result<(), DomainError> { - // Count total blobs for progress tracking. - let total: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs") - .fetch_one(pool.as_ref()) - .await - .unwrap_or(0); - - { - let mut s = state.write().await; - s.status = MigrationStatus::Running; - s.total_blobs = total as u64; - s.migrated_blobs = 0; - s.migrated_bytes = 0; - s.failed_blobs.clear(); - s.started_at = Some(chrono::Utc::now()); - s.completed_at = None; - } - - // Stream all hashes+sizes with a cursor. - let mut rows = - sqlx::query_as::<_, (String, i64)>("SELECT hash, size FROM storage.blobs ORDER BY hash") - .fetch(pool.as_ref()); - - // Collect all hashes first to avoid holding the cursor across awaits. - let mut work: Vec<(String, i64)> = Vec::with_capacity(total as usize); - while let Some(row) = rows.next().await { - match row { - Ok(r) => work.push(r), - Err(e) => { - tracing::warn!("Error fetching blob row during migration: {}", e); - } - } - } - - // Process in parallel chunks. - let results = futures::stream::iter(work.into_iter().map(|(hash, size)| { - let src = source.clone(); - let tgt = target.clone(); - let st = state.clone(); - async move { - // Check if we should keep running. - { - let s = st.read().await; - if s.status != MigrationStatus::Running { - return; - } - } - - // Skip if already in target. - match tgt.blob_exists(&hash).await { - Ok(true) => { - let mut s = st.write().await; - s.migrated_blobs += 1; - s.migrated_bytes += size as u64; - return; - } - Ok(false) => {} - Err(e) => { - tracing::warn!("blob_exists check failed for {}: {}", hash, e); - } - } - - // Copy: stream from source → temp file → put into target. - if let Err(e) = copy_blob(&src, &tgt, &hash).await { - tracing::warn!("Failed to migrate blob {}: {}", hash, e); - let mut s = st.write().await; - s.failed_blobs.push(hash); - return; - } - - let mut s = st.write().await; - s.migrated_blobs += 1; - s.migrated_bytes += size as u64; - } - })) - .buffer_unordered(concurrency) - .collect::>() - .await; - - drop(results); - - // Finalize state. - let mut s = state.write().await; - if s.status == MigrationStatus::Running { - if s.failed_blobs.is_empty() { - s.status = MigrationStatus::Completed; - } else { - s.status = MigrationStatus::Failed; - } - s.completed_at = Some(chrono::Utc::now()); - } - - tracing::info!( - "Migration finished: {}/{} blobs, {} failures", - s.migrated_blobs, - s.total_blobs, - s.failed_blobs.len() - ); - - Ok(()) -} - -/// Copy a single blob: stream from source → spool to temp file → put_blob into target. -async fn copy_blob( - source: &Arc, - target: &Arc, - hash: &str, -) -> Result<(), DomainError> { - use tokio::io::AsyncWriteExt; - - // Create a temp file to spool content. - let tmp_dir = std::env::temp_dir().join("oxicloud-migration"); - tokio::fs::create_dir_all(&tmp_dir).await.map_err(|e| { - DomainError::internal_error("Migration", format!("Failed to create temp dir: {}", e)) - })?; - - let tmp_path = tmp_dir.join(format!("{}.tmp", hash)); - - // Stream from source. - let stream = source.get_blob_stream(hash).await?; - - // Write to temp file. - let mut file = tokio::fs::File::create(&tmp_path).await.map_err(|e| { - DomainError::internal_error("Migration", format!("Failed to create temp file: {}", e)) - })?; - - let mut stream = std::pin::pin!(stream); - while let Some(chunk) = stream.next().await { - let bytes = chunk.map_err(|e| { - DomainError::internal_error("Migration", format!("Stream error: {}", e)) - })?; - file.write_all(&bytes) - .await - .map_err(|e| DomainError::internal_error("Migration", format!("Write error: {}", e)))?; - } - file.flush() - .await - .map_err(|e| DomainError::internal_error("Migration", format!("Flush error: {}", e)))?; - drop(file); - - // Put into target. - target.put_blob(hash, &tmp_path).await?; - - // Clean up temp file. - let _ = tokio::fs::remove_file(&tmp_path).await; - - Ok(()) -} - -/// Verify migration integrity by comparing blob counts and sampling random hashes. -pub async fn verify_migration( - target: Arc, - pool: Arc, - sample_size: usize, -) -> Result { - // 1. Count blobs in PG. - let pg_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM storage.blobs") - .fetch_one(pool.as_ref()) - .await - .unwrap_or(0); - - // 2. Verify sample of blobs exist in target. - 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.as_ref()) - .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) => { - // Verify size matches. - 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(VerificationResult { - pg_blob_count: pg_count as u64, - sample_checked: sample_rows.len() as u64, - missing_in_target: missing, - size_mismatches, - passed, - }) -} - -/// Result of a post-migration integrity check. -#[derive(Debug, Clone, Serialize, serde::Deserialize)] -pub struct VerificationResult { - pub pg_blob_count: u64, - pub sample_checked: u64, - pub missing_in_target: Vec, - pub size_mismatches: Vec, - pub passed: bool, -} diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index 457e0ef1..1565e65f 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -25,8 +25,6 @@ pub mod local_blob_backend; pub mod local_fs_mount_provider; pub mod login_lockout_service; pub mod media_metadata_service; -pub mod migration_blob_backend; -pub mod migration_job; pub mod mock_email_sender; pub mod mount_provider_factory; pub mod nextcloud_chunked_upload_service; @@ -46,6 +44,7 @@ pub mod s3_blob_backend; pub mod search_index; pub mod share_unlock_cookie; pub mod smtp_email_sender; +pub mod storage_migration_service; pub mod thumbnail_service; #[cfg(test)] mod thumbnail_service_test; diff --git a/src/infrastructure/services/storage_migration_service.rs b/src/infrastructure/services/storage_migration_service.rs new file mode 100644 index 00000000..4f8b0979 --- /dev/null +++ b/src/infrastructure/services/storage_migration_service.rs @@ -0,0 +1,486 @@ +//! Storage-backend migration as a recoverable-run tenant (Part 2 engine). +//! +//! Iterates `storage.blobs` and copies each byte payload from the SOURCE +//! backend (whatever the app booted with) to the TARGET backend +//! (whatever the current admin storage-settings config describes). +//! Both legacy whole-file blobs AND CDC chunk blobs are covered by the +//! single walk — they share `storage.blobs` as their physical registry +//! (see memory `project_cdc_dual_storage_registries`). +//! `storage.chunk_manifests` is pure PG state, holds no backend bytes, +//! and needs no migration. +//! +//! Retires the in-memory `Arc>` + one-shot +//! `tokio::spawn` in `migration_job.rs`. The recoverable engine +//! provides cursor persistence, cooperative cancel, boot-time crash +//! recovery, and the uniform `/api/admin/jobs/*` admin surface. +//! +//! ### Restart survival +//! +//! Cursor + per-blob failure findings are persisted after every batch. +//! On restart the boot-time sweep flips any abandoned `Running` row to +//! `Paused`; a subsequent admin trigger resumes from the persisted +//! cursor via `run_or_resume`. At most one batch of already-copied +//! blobs replays, and the `target.blob_exists` short-circuit makes +//! even that replay effectively free. +//! +//! ### Design notes +//! +//! * **Cursor.** UTF-8 hex of the last-processed blob hash (64 chars). +//! Natural lex order matches `ORDER BY hash ASC`. Same encoding +//! `blobs_consistency` uses. +//! * **Target resolution.** Rebuilt at the START of every fresh or +//! resumed run via `StorageSettingsService::build_effective_backend`. +//! Held for the duration of the run; a mid-run settings change is +//! ignored until the next run. On resume the admin may have paused +//! *specifically* to fix a broken target config, so we re-derive +//! rather than pin. +//! * **Per-blob failures don't fail the run.** Each failure records a +//! `migration_failed` finding (severity `data_loss` — the bytes +//! didn't cross) and the walk continues. A run that completes with +//! zero findings is proof the target has every blob. +//! * **Skip already-present blobs.** `target.blob_exists(hash)` before +//! the copy — makes cheap re-runs safe and lets a paused run resume +//! without redoing bytes. +//! * **No per-batch concurrency knob.** The old code buffered N copies +//! in parallel. Sequential is easier to reason about with cooperative +//! 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::sync::Arc; + +use async_trait::async_trait; +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::errors::DomainError; +use crate::infrastructure::scheduler::{ + JobRegistry, JobRunArgs, JobStore, JobStoreProvider, RecoverableJobHandler, RunOutcome, + RunStatus, record_or_log, +}; + +pub const STORAGE_MIGRATION_JOB_NAME: &str = "storage_migration"; + +/// 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 +/// checkpoint per ~hundred blobs is fine, and the cancel-poll comes +/// every 100 rows too. Match `blobs_consistency` for consistency. +const BATCH_SIZE: i64 = 100; + +pub struct StorageMigrationService { + pool: Arc, + source: Arc, + storage_settings: Arc, +} + +impl StorageMigrationService { + pub fn new( + pool: Arc, + source: Arc, + storage_settings: Arc, + ) -> Self { + Self { + pool, + source, + storage_settings, + } + } + + /// Chainable self-registration — mirrors the `*_consistency` + /// tenants. 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 StorageMigrationService { + fn name(&self) -> &str { + STORAGE_MIGRATION_JOB_NAME + } + + /// Definitive count — one row per blob. `SELECT COUNT(*) FROM + /// storage.blobs` on a modern PG is a sub-second index-only scan + /// even at millions of 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::migration", + event = "storage_migration.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 { + // 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" + ); + return RunOutcome::Failed { + message: + "target equals source; change storage settings before triggering a migration" + .to_string(), + }; + } + Ok(false) => {} + Err(e) => { + return RunOutcome::Failed { + message: format!("identity check: {e}"), + }; + } + } + + // Resolve target at run start. + let target = match self.storage_settings.build_effective_backend().await { + Ok(t) => t, + Err(e) => { + return RunOutcome::Failed { + message: format!("resolve target backend: {e}"), + }; + } + }; + if let Err(e) = target.initialize().await { + return RunOutcome::Failed { + message: format!("target backend init: {e}"), + }; + } + + let source_kind = self.source.backend_type(); + let target_kind = target.backend_type(); + tracing::info!( + 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}" + ); + + // Cursor = the last-visited blob hash, UTF-8-encoded. On resume + // walk `WHERE hash > $cursor`. `None` / empty = start from the + // smallest hash. Same shape `blobs_consistency` uses. + 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) => { + return RunOutcome::Failed { + message: format!("invalid cursor: not valid UTF-8: {e}"), + }; + } + }, + }; + + let mut copied_count = 0u64; + let mut skipped_count = 0u64; + let mut failed_count = 0u64; + let mut source_missing_count = 0u64; + + loop { + // Cooperative cancel poll between batches. + match store.status().await { + Ok(RunStatus::CancelRequested) => { + tracing::info!( + target: "oxicloud::migration", + event = "storage_migration.cancelled", + run_id = %store.run_id(), + copied = copied_count, + skipped = skipped_count, + failed = failed_count, + source_missing = source_missing_count, + "storage_migration cancelled cooperatively, pausing" + ); + return RunOutcome::Paused { + cursor: cursor + .as_ref() + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(), + }; + } + Ok(_) => {} + Err(e) => { + return RunOutcome::Failed { + message: format!("status poll: {e}"), + }; + } + } + + // Fetch the next batch. `hash > $1` keyset pagination on + // the PK; index-only scan. + let rows: Vec<(String, i64)> = match sqlx::query_as( + r#" + SELECT hash, size + 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) => { + return RunOutcome::Failed { + message: format!("batch fetch: {e}"), + }; + } + }; + + if rows.is_empty() { + tracing::info!( + target: "oxicloud::migration", + event = "storage_migration.completed", + run_id = %store.run_id(), + copied = copied_count, + skipped = skipped_count, + failed = failed_count, + source_missing = source_missing_count, + "storage_migration completed" + ); + return RunOutcome::Completed; + } + + for (hash, size) in &rows { + // Probe SOURCE first — without this a run would + // silently "succeed" against a source that's missing + // blobs the DB expects, and the audit intent of the + // walk is lost (relevant on any post-migration state + // where target may already have every blob). A + // missing-on-source blob is a real data-loss + // condition; record it and move on — we never + // "copy" from nothing. + match self.source.blob_exists(hash).await { + Ok(true) => {} + Ok(false) => { + source_missing_count += 1; + tracing::warn!( + target: "oxicloud::migration", + event = "storage_migration.source_missing", + run_id = %store.run_id(), + hash = %hash, + source = source_kind, + "blob absent from source; recording data-loss finding, no copy" + ); + record_or_log( + store, + STORAGE_MIGRATION_JOB_NAME, + "source_missing", + "data_loss", + None, + serde_json::json!({ + "hash": hash, + "size": size, + "source": source_kind, + "target": target_kind, + }), + ) + .await; + continue; + } + Err(e) => { + // Transient probe failure on source is NOT a + // finding — treat like a network blip. + // Skipping this row on this run; a re-run + // will re-probe. If the failure is + // persistent, `blobs_consistency` catches + // it. + tracing::warn!( + target: "oxicloud::migration", + event = "storage_migration.source_probe_error", + run_id = %store.run_id(), + hash = %hash, + error = %e, + "source blob_exists probe failed; skipping this row" + ); + continue; + } + } + + // Skip when the target already has it — supports + // idempotent resume and cheap re-runs against a + // partially-migrated target. + match target.blob_exists(hash).await { + Ok(true) => { + skipped_count += 1; + continue; + } + Ok(false) => {} + Err(e) => { + tracing::warn!( + target: "oxicloud::migration", + event = "storage_migration.blob_exists_error", + run_id = %store.run_id(), + hash = %hash, + error = %e, + "blob_exists probe on target failed; attempting copy anyway" + ); + } + } + + match copy_blob(self.source.as_ref(), target.as_ref(), hash).await { + Ok(()) => { + copied_count += 1; + } + Err(e) => { + failed_count += 1; + tracing::warn!( + target: "oxicloud::migration", + event = "storage_migration.blob_failed", + run_id = %store.run_id(), + hash = %hash, + error = %e, + "failed to migrate blob; recording finding, continuing" + ); + // resource_id stays None — blob hash isn't a + // UUID. Real identifier lives in `detail.hash` + // where the admin UI reads it. + record_or_log( + store, + STORAGE_MIGRATION_JOB_NAME, + "migration_failed", + "data_loss", + None, + serde_json::json!({ + "hash": hash, + "size": size, + "source": source_kind, + "target": target_kind, + "error": e.to_string(), + }), + ) + .await; + } + } + } + + // Advance cursor + checkpoint. `delta_count` counts WORK + // ATTEMPTED (copied + skipped + failed), not successful + // copies alone — otherwise the progress bar stalls whenever + // a batch is dominated by already-present blobs, which is + // exactly the case on a resume. + 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 { + return RunOutcome::Failed { + message: format!("checkpoint: {e}"), + }; + } + + if (rows.len() as i64) < BATCH_SIZE { + tracing::info!( + target: "oxicloud::migration", + event = "storage_migration.completed", + run_id = %store.run_id(), + copied = copied_count, + skipped = skipped_count, + failed = failed_count, + source_missing = source_missing_count, + "storage_migration completed" + ); + return RunOutcome::Completed; + } + } + } +} + +/// 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' +/// `put_blob` want a path they can `rename(2)` or multi-part upload +/// from, not an in-memory buffer. The temp file lives in +/// `std::env::temp_dir()/oxicloud-migration/{hash}.tmp` and is +/// removed on success (best-effort on the failure paths — the OS +/// cleans up on reboot). +async fn copy_blob( + source: &dyn BlobStorageBackend, + target: &dyn BlobStorageBackend, + hash: &str, +) -> Result<(), DomainError> { + let tmp_dir = std::env::temp_dir().join("oxicloud-migration"); + tokio::fs::create_dir_all(&tmp_dir).await.map_err(|e| { + DomainError::internal_error( + "StorageMigration", + format!("create temp dir {}: {e}", tmp_dir.display()), + ) + })?; + let tmp_path = tmp_dir.join(format!("{hash}.tmp")); + + if let Err(e) = write_source_to_tmp(source, hash, &tmp_path).await { + let _ = tokio::fs::remove_file(&tmp_path).await; + return Err(e); + } + + let put_result = target.put_blob(hash, &tmp_path).await; + let _ = tokio::fs::remove_file(&tmp_path).await; + put_result.map(|_bytes_written| ()) +} + +async fn write_source_to_tmp( + source: &dyn BlobStorageBackend, + hash: &str, + tmp_path: &Path, +) -> Result<(), DomainError> { + use tokio::io::AsyncWriteExt; + + let stream = source.get_blob_stream(hash).await?; + let mut file = tokio::fs::File::create(tmp_path).await.map_err(|e| { + DomainError::internal_error( + "StorageMigration", + format!("create temp file {}: {e}", tmp_path.display()), + ) + })?; + let mut stream = std::pin::pin!(stream); + while let Some(chunk) = stream.next().await { + let bytes = chunk.map_err(|e| { + DomainError::internal_error("StorageMigration", format!("source stream read: {e}")) + })?; + file.write_all(&bytes).await.map_err(|e| { + DomainError::internal_error("StorageMigration", format!("temp file write: {e}")) + })?; + } + file.flush() + .await + .map_err(|e| DomainError::internal_error("StorageMigration", format!("temp flush: {e}")))?; + Ok(()) +} diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index 77f8e7f5..eb172bda 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -24,9 +24,13 @@ use crate::application::dtos::settings_dto::{ use crate::application::dtos::user_dto::{AdminUserSummaryDto, UserDto}; use crate::application::ports::authorization_ports::AuthorizationEngine; use crate::application::ports::plugin_ports::{LogQuery, PluginManagementPort, PluginMgmtError}; +// JobStoreProvider is used only by the storage-migration shims below, +// but the compiler needs the trait in scope for method resolution on +// the concrete `PgJobStoreProvider` that lives on `AppState`. use crate::common::di::AppState; use crate::domain::repositories::drive_repository::DriveRepository; use crate::domain::services::authorization::{Resource, Subject}; +use crate::infrastructure::scheduler::JobStoreProvider; use crate::interfaces::api::handlers::dedup_handler::{get_stats, recalculate_stats}; use crate::interfaces::api::handlers::search_handler::clear_search_cache; use crate::interfaces::errors::AppError; @@ -70,12 +74,17 @@ pub fn admin_routes() -> Router> { .route("/settings/storage", get(get_storage_settings)) .route("/settings/storage", put(save_storage_settings)) .route("/settings/storage/test", post(test_storage_connection)) - // Storage migration + // Storage migration — thin shims over the recoverable-run + // engine (job_name = "storage_migration"). Retained under + // /storage/migration/* until the admin UI is rewired to + // /api/admin/jobs/storage_migration/*; both paths route to + // the same underlying JobRegistry dispatch. The old /complete + // endpoint is retired — a finished run is a Completed row, + // there's nothing to acknowledge. .route("/storage/migration", get(get_migration_status)) .route("/storage/migration/start", post(start_migration)) .route("/storage/migration/pause", post(pause_migration)) .route("/storage/migration/resume", post(resume_migration)) - .route("/storage/migration/complete", post(complete_migration)) .route("/storage/migration/verify", post(verify_migration)) // Encryption key generation .route( @@ -361,7 +370,15 @@ async fn test_storage_connection( // Storage migration handlers // ───────────────────────────────────────────────────── -/// GET /api/admin/storage/migration — current migration progress +/// GET /api/admin/storage/migration — current migration progress. +/// +/// Shim over the recoverable-run engine: reads the latest +/// `storage_migration` run from `jobs.recoverable_runs` (via the +/// `JobStoreProvider`) and projects it into the legacy +/// `MigrationStateDto` shape the admin storage tab expects. When no +/// run has ever been triggered the response is an empty "idle" DTO — +/// same behaviour the old in-memory `MigrationState::default()` +/// produced. #[utoipa::path( get, path = "/api/admin/storage/migration", @@ -376,17 +393,59 @@ async fn test_storage_connection( pub async fn get_migration_status( State(state): State>, ) -> Result { - let s = state.migration_state.read().await; - Ok(Json(migration_state_to_dto(&s))) + use crate::infrastructure::services::storage_migration_service::STORAGE_MIGRATION_JOB_NAME; + + let provider = state.core.job_store_provider.clone(); + let latest = provider + .list_runs(STORAGE_MIGRATION_JOB_NAME, 1) + .await + .map_err(AppError::from)? + .into_iter() + .next(); + + let Some(run) = latest else { + return Ok(Json(idle_migration_dto())); + }; + + // Failed blobs are stored as findings, kind = "migration_failed". + // Pull up to a reasonable ceiling — the DTO ships the full list, + // and the admin UI truncates its own display. + let findings = provider + .list_findings(run.id, 500, 0) + .await + .map_err(AppError::from)?; + let failed_blobs: Vec = findings + .into_iter() + .filter(|f| f.kind == "migration_failed") + .filter_map(|f| { + f.detail + .get("hash") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()) + }) + .collect(); + + Ok(Json(run_to_migration_dto(&run, failed_blobs))) } -/// POST /api/admin/storage/migration/start — begin background migration +/// POST /api/admin/storage/migration/start — begin background migration. +/// +/// Shim that forwards to `JobRegistry::trigger("storage_migration", +/// ...)`. `run_or_resume` (the RecoverableAdapter's inner dispatch) +/// resumes a Paused run or starts a fresh one — one endpoint covers +/// both. Exclusivity is enforced at the DB layer (the partial unique +/// index on `jobs.recoverable_runs`), so a second concurrent trigger +/// is a no-op that returns the existing run. +/// +/// `StartMigrationDto.concurrency` is currently ignored — the +/// recoverable copy loop runs sequentially. Kept in the DTO for +/// wire-compat with the admin UI; will be honoured if a concurrency +/// knob is added later. #[utoipa::path( post, path = "/api/admin/storage/migration/start", responses( (status = 200, description = "Migration started"), - (status = 400, description = "Migration already running"), (status = 401, description = "Unauthorized"), (status = 403, description = "Admin required") ), @@ -395,73 +454,23 @@ pub async fn get_migration_status( )] pub async fn start_migration( State(state): State>, - Json(dto): Json, + Json(_dto): Json, ) -> Result { - use crate::infrastructure::services::migration_blob_backend::MigrationStatus; - - // Check not already running. - { - let s = state.migration_state.read().await; - if s.status == MigrationStatus::Running { - return Err(AppError::bad_request("A migration is already running")); - } - } - - let pool = state - .db_pool - .clone() - .ok_or_else(|| AppError::internal_error("Database not available"))?; - - let source = state.core.dedup_service.backend().clone(); - let svc = state - .storage_settings_service - .as_ref() - .ok_or_else(|| AppError::internal_error("Storage settings service not available"))?; - - // Build target backend from saved settings. - let effective = svc - .load_effective_storage_config() - .await - .map_err(|e| AppError::internal_error(format!("Failed to load storage config: {}", e)))?; - - let target = build_backend_from_config(&effective) - .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 concurrency = dto.concurrency.unwrap_or(4).clamp(1, 16); - let migration_state = state.migration_state.clone(); - - // Spawn the background migration job. - tokio::spawn(async move { - if let Err(e) = crate::infrastructure::services::migration_job::run_migration( - source, - target, - pool, - migration_state, - concurrency, - ) - .await - { - tracing::error!("Migration job error: {}", e); - } - }); - - Ok(( - StatusCode::OK, - Json(serde_json::json!({ "message": "Migration started" })), - )) + trigger_storage_migration(state).await } -/// POST /api/admin/storage/migration/pause — pause running migration +/// POST /api/admin/storage/migration/pause — pause a running migration. +/// +/// Shim over cooperative cancel: flips the run row's status to +/// `CancelRequested`; the recoverable handler polls between batches +/// and returns `Paused` at the next boundary. If nothing is running, +/// returns 200 with `paused: false` — matches the "no-op is fine" +/// contract of `/api/admin/jobs/{name}/cancel`. #[utoipa::path( post, path = "/api/admin/storage/migration/pause", responses( - (status = 200, description = "Migration paused"), - (status = 400, description = "No running migration"), + (status = 200, description = "Pause signalled (or no-op)"), (status = 401, description = "Unauthorized"), (status = 403, description = "Admin required") ), @@ -471,26 +480,45 @@ pub async fn start_migration( pub async fn pause_migration( State(state): State>, ) -> Result { - use crate::infrastructure::services::migration_blob_backend::MigrationStatus; + use crate::infrastructure::services::storage_migration_service::STORAGE_MIGRATION_JOB_NAME; + + tracing::info!( + target: "audit", + event = "storage_migration.pause_requested", + "👮🏻‍♂️ Admin requested storage_migration pause" + ); + + let flipped = state + .core + .job_store_provider + .request_cancel(STORAGE_MIGRATION_JOB_NAME) + .await + .map_err(AppError::from)?; - let mut s = state.migration_state.write().await; - if s.status != MigrationStatus::Running { - return Err(AppError::bad_request("No running migration to pause")); - } - s.status = MigrationStatus::Paused; Ok(( StatusCode::OK, - Json(serde_json::json!({ "message": "Migration paused" })), + Json(serde_json::json!({ + "paused": flipped.is_some(), + "run_id": flipped, + "message": if flipped.is_some() { + "Pause requested — handler will yield at the next batch boundary" + } else { + "No running migration to pause" + }, + })), )) } -/// POST /api/admin/storage/migration/resume — resume paused migration +/// POST /api/admin/storage/migration/resume — resume a paused migration. +/// +/// Same underlying trigger as `/start`: `run_or_resume` inspects the +/// latest row and picks Fresh / Resume / AlreadyActive at dispatch +/// time. Kept as a distinct endpoint for wire-compat. #[utoipa::path( post, path = "/api/admin/storage/migration/resume", responses( - (status = 200, description = "Migration resumed"), - (status = 400, description = "No paused migration"), + (status = 200, description = "Migration resumed (or already running)"), (status = 401, description = "Unauthorized"), (status = 403, description = "Admin required") ), @@ -500,59 +528,15 @@ pub async fn pause_migration( pub async fn resume_migration( State(state): State>, ) -> Result { - use crate::infrastructure::services::migration_blob_backend::MigrationStatus; - - // Set status back to Running — the background task checks on each blob. - let mut s = state.migration_state.write().await; - if s.status != MigrationStatus::Paused { - return Err(AppError::bad_request("No paused migration to resume")); - } - s.status = MigrationStatus::Running; - Ok(( - StatusCode::OK, - Json(serde_json::json!({ "message": "Migration resumed" })), - )) + trigger_storage_migration(state).await } -/// POST /api/admin/storage/migration/complete — finalize migration -#[utoipa::path( - post, - path = "/api/admin/storage/migration/complete", - responses( - (status = 200, description = "Migration finalized"), - (status = 400, description = "Migration not completed"), - (status = 401, description = "Unauthorized"), - (status = 403, description = "Admin required") - ), - security(("bearerAuth" = [])), - tag = "admin" -)] -pub async fn complete_migration( - State(state): State>, -) -> Result { - use crate::infrastructure::services::migration_blob_backend::MigrationStatus; - - let s = state.migration_state.read().await; - if s.status != MigrationStatus::Completed { - return Err(AppError::bad_request( - "Migration must be completed (100%) before finalizing", - )); - } - drop(s); - - // Mark as idle — the admin has acknowledged completion. - let mut s = state.migration_state.write().await; - s.status = MigrationStatus::Idle; - - Ok(( - StatusCode::OK, - Json( - serde_json::json!({ "message": "Migration finalized. Restart the server to use the new backend." }), - ), - )) -} - -/// POST /api/admin/storage/migration/verify — run integrity check +/// 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", @@ -579,12 +563,9 @@ pub async fn verify_migration( .as_ref() .ok_or_else(|| AppError::internal_error("Storage settings service not available"))?; - let effective = svc - .load_effective_storage_config() + let target = svc + .build_effective_backend() .await - .map_err(|e| AppError::internal_error(format!("Failed to load storage config: {}", e)))?; - - let target = build_backend_from_config(&effective) .map_err(|e| AppError::internal_error(format!("Failed to build target backend: {}", e)))?; target .initialize() @@ -593,38 +574,176 @@ pub async fn verify_migration( let sample_size = dto.sample_size.unwrap_or(100).clamp(1, 1000); - let result = - crate::infrastructure::services::migration_job::verify_migration(target, pool, sample_size) - .await - .map_err(|e| AppError::internal_error(format!("Verification failed: {}", e)))?; + 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)) } -/// Helper: convert MigrationState to DTO for JSON serialization. -fn migration_state_to_dto( - s: &crate::infrastructure::services::migration_blob_backend::MigrationState, -) -> MigrationStateDto { - let throughput = match (s.started_at, s.migrated_bytes) { - (Some(start), bytes) if bytes > 0 => { - let elapsed = chrono::Utc::now() - .signed_duration_since(start) - .num_seconds() - .max(1) as f64; - Some(bytes as f64 / elapsed) +/// Shared body for `start` / `resume` — both funnel through +/// `run_or_resume` via `JobRegistry::trigger`. Detaches into a +/// `tokio::spawn` so the HTTP response returns immediately — same +/// rationale as `trigger_job` above (browser timeout mid-await would +/// desync `current_run_start` from the actually-running task). The +/// admin UI polls `GET /storage/migration` for progress; the trigger +/// itself is fire-and-forget. +async fn trigger_storage_migration( + state: Arc, +) -> Result { + use crate::infrastructure::scheduler::JobRunArgs; + use crate::infrastructure::services::storage_migration_service::STORAGE_MIGRATION_JOB_NAME; + + tracing::info!( + target: "audit", + event = "storage_migration.trigger_requested", + "👮🏻‍♂️ Admin triggered storage_migration" + ); + + let registry = state.core.job_registry.clone(); + tokio::spawn(async move { + registry + .trigger(STORAGE_MIGRATION_JOB_NAME, &JobRunArgs::default()) + .await; + }); + + Ok(( + StatusCode::ACCEPTED, + Json(serde_json::json!({ + "message": "Migration dispatched — poll GET /api/admin/storage/migration for status", + "detached": true, + })), + ) + .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()); + } + } } - _ => None, + } + + 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 { + status: "idle".to_string(), + total_blobs: 0, + migrated_blobs: 0, + // `migrated_bytes` and `throughput_bytes_per_sec` are no + // longer tracked — the recoverable engine bumps + // `stats.scanned_count` (a blob-count aggregator), not a + // bytes counter. The admin UI keeps these fields for + // wire-compat; they read 0 / null. + migrated_bytes: 0, + failed_blobs: Vec::new(), + started_at: None, + completed_at: None, + throughput_bytes_per_sec: None, + } +} + +/// Project a recoverable `RunSummary` into the admin UI's +/// `MigrationStateDto`. Byte-counter fields are always 0 / None — +/// see `idle_migration_dto`'s comment. +fn run_to_migration_dto( + run: &crate::infrastructure::scheduler::RunSummary, + failed_blobs: Vec, +) -> MigrationStateDto { + use crate::infrastructure::scheduler::RunStatus; + + // Fold CancelRequested into "paused" — from the admin UI's + // point of view a cancel-in-flight is the "waiting for the + // handler to yield" state. Same visual affordance as Paused. + let status = match run.status { + RunStatus::Running => "running", + RunStatus::Paused => "paused", + RunStatus::CancelRequested => "paused", + RunStatus::Completed => "completed", + RunStatus::Failed => "failed", + } + .to_string(); + + let (total_blobs, migrated_blobs) = match run.progress.as_ref() { + Some(p) => (p.total, p.scanned), + None => ( + 0, + run.stats + .get("scanned_count") + .and_then(|v| v.as_u64()) + .unwrap_or(0), + ), }; MigrationStateDto { - status: format!("{:?}", s.status).to_lowercase(), - total_blobs: s.total_blobs, - migrated_blobs: s.migrated_blobs, - migrated_bytes: s.migrated_bytes, - failed_blobs: s.failed_blobs.clone(), - started_at: s.started_at.map(|d| d.to_rfc3339()), - completed_at: s.completed_at.map(|d| d.to_rfc3339()), - throughput_bytes_per_sec: throughput, + status, + total_blobs, + migrated_blobs, + migrated_bytes: 0, + failed_blobs, + started_at: Some(run.started_at.to_rfc3339()), + completed_at: run.completed_at.map(|d| d.to_rfc3339()), + throughput_bytes_per_sec: None, } } @@ -652,34 +771,6 @@ pub async fn generate_encryption_key() -> Result { }))) } -/// Helper: build a BlobStorageBackend from StorageConfig. -fn build_backend_from_config( - config: &crate::common::config::StorageConfig, -) -> Result< - std::sync::Arc, - String, -> { - match config.backend { - crate::common::config::StorageBackendType::Local => Ok(std::sync::Arc::new( - crate::infrastructure::services::local_blob_backend::LocalBlobBackend::new( - std::path::Path::new(&config.root_dir), - ), - )), - crate::common::config::StorageBackendType::S3 => { - let s3 = config.s3.as_ref().ok_or("S3 config missing")?; - Ok(std::sync::Arc::new( - crate::infrastructure::services::s3_blob_backend::S3BlobBackend::new(s3), - )) - } - crate::common::config::StorageBackendType::Azure => { - let az = config.azure.as_ref().ok_or("Azure config missing")?; - Ok(std::sync::Arc::new( - crate::infrastructure::services::azure_blob_backend::AzureBlobBackend::new(az), - )) - } - } -} - // ============================================================================ // Dashboard / Stats // ============================================================================ @@ -2159,6 +2250,40 @@ pub async fn trigger_job( force: query.force, deep: query.deep, }; + + // Jobs that can run for hours (storage_migration, future + // reextract_*) are detached: `tokio::spawn` the trigger so the + // HTTP request returns immediately. Without this, browser HTTP + // timeouts drop the request future mid-await → the SemaphorePermit + // gets released while the spawned handler task keeps running → + // `current_run_start` goes stale → a second click enters the + // "already_running" short-circuit and CLEARS the in-memory state + // even though the original task is still copying blobs → the + // Cancel button hides because `job.running = false`. Detaching + // keeps the permit + `current_run_start` scoped to the actual + // handler-task lifetime. + // + // Fast-completing jobs (consistency checks, batch coordinator) + // stay inline so the operator sees the outcome envelope. + if is_detached_job(&name) { + let name_clone = name.clone(); + let registry = state.core.job_registry.clone(); + tokio::spawn(async move { + registry.trigger(&name_clone, &args).await; + }); + return ( + StatusCode::ACCEPTED, + Json(serde_json::json!({ + "ok": true, + "dispatched": true, + "detached": true, + "name": name, + "message": "Dispatched — poll /runs for status", + })), + ) + .into_response(); + } + match state.core.job_registry.trigger(&name, &args).await { Some(outcome) => ( StatusCode::OK, @@ -2176,6 +2301,15 @@ pub async fn trigger_job( } } +/// Jobs that MUST be dispatched with `tokio::spawn` because they run +/// long enough to outlast an HTTP request timeout. Kept as a small +/// hardcoded allowlist (rather than a flag on `JobEntry`) until +/// there's a second long-running tenant that justifies the plumbing. +/// See the comment in `trigger_job` for why detach matters. +fn is_detached_job(name: &str) -> bool { + matches!(name, "storage_migration") +} + /// `POST /api/admin/jobs/{name}/cancel` — cooperative cancel of the /// currently-running recoverable run for `{name}`. /// diff --git a/src/interfaces/api/mod.rs b/src/interfaces/api/mod.rs index 64d7308d..e9c510d7 100644 --- a/src/interfaces/api/mod.rs +++ b/src/interfaces/api/mod.rs @@ -225,7 +225,6 @@ 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::complete_migration, handlers::admin_handler::verify_migration, handlers::admin_handler::generate_encryption_key, // JobRegistry admin surface — production, always-on,