cjw/Oxicloud
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

feat(storage-migration): move storage mig. to recoverable job

Browse Source
This commit is contained in:
Edouard Vanbelle
2026-07-30 21:21:26 +02:00
parent 0c6a33c721
commit 82a50936d4
12 changed files with 1436 additions and 745 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/application/dtos/settings_dto.rs
+38 -1
View File
@@ -191,13 +191,50 @@ pub struct TestStorageConnectionDto {
pub s3_force_path_style: Option<bool>,
}
/// 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<u64>,
/// 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<bool>,
/// 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<String>,
#[serde(skip_serializing_if = "Option::is_none")]
pub bytes_written: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub bytes_read: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub roundtrip_elapsed_ms: Option<u64>,
#[serde(skip_serializing_if = "Option::is_none")]
pub cleanup_ok: Option<bool>,
}
// ============================================================================
src/application/services/storage_settings_service.rs
+344 -12
View File
@@ -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:<missing-config>".to_string(),
},
StorageBackendType::Azure => match config.azure.as_ref() {
Some(az) => format!(
"azure:{}/{}",
az.account_name.as_str(),
az.container.as_str(),
),
None => "azure:<missing-config>".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<bool, DomainError> {
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<Arc<dyn BlobStorageBackend>, 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<StorageConfig, DomainError> {
let db: HashMap<String, String> = 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<StorageTestResultDto, DomainError> {
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<u8> = 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}"
)
},
)
}
src/common/di.rs
+25 -4
View File
@@ -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<Arc<dyn crate::application::ports::plugin_ports::PluginManagementPort>>,
pub storage_settings_service: Option<Arc<StorageSettingsService>>,
pub migration_state: Arc<tokio::sync::RwLock<MigrationState>>,
pub trash_service: Option<Arc<TrashService>>,
pub share_service: Option<Arc<ShareService>>,
pub share_browse_service: Option<Arc<ShareBrowseService>>,
src/infrastructure/services/migration_blob_backend.rs
-273
View File
@@ -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<String>,
pub started_at: Option<DateTime<Utc>>,
pub completed_at: Option<DateTime<Utc>>,
}
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<dyn BlobStorageBackend>,
target: Arc<dyn BlobStorageBackend>,
state: Arc<RwLock<MigrationState>>,
}
impl MigrationBlobBackend {
pub fn new(
source: Arc<dyn BlobStorageBackend>,
target: Arc<dyn BlobStorageBackend>,
state: Arc<RwLock<MigrationState>>,
) -> Self {
Self {
source,
target,
state,
}
}
pub fn state(&self) -> &Arc<RwLock<MigrationState>> {
&self.state
}
pub fn source(&self) -> &Arc<dyn BlobStorageBackend> {
&self.source
}
pub fn target(&self) -> &Arc<dyn BlobStorageBackend> {
&self.target
}
}
/// Boxed future alias (same as in the trait module).
type BoxFut<'a, T> = Pin<Box<dyn Future<Output = T> + 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<u64, DomainError>> {
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<u64, DomainError>> {
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<u64, DomainError>> {
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<BlobStream, DomainError>> {
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<u64>,
) -> BoxFut<'_, Result<BlobStream, DomainError>> {
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<bool, DomainError>> {
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<u64, DomainError>> {
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<StorageHealthStatus, DomainError>> {
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<PathBuf> {
// 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<String>,
_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",
))
})
}
}
src/infrastructure/services/migration_job.rs
-240
View File
@@ -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<dyn BlobStorageBackend>,
target: Arc<dyn BlobStorageBackend>,
pool: Arc<PgPool>,
state: Arc<RwLock<MigrationState>>,
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::<Vec<()>>()
.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<dyn BlobStorageBackend>,
target: &Arc<dyn BlobStorageBackend>,
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<dyn BlobStorageBackend>,
pool: Arc<PgPool>,
sample_size: usize,
) -> Result<VerificationResult, DomainError> {
// 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<String>,
pub size_mismatches: Vec<String>,
pub passed: bool,
}
src/infrastructure/services/mod.rs
+1 -2
View File
@@ -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;
src/infrastructure/services/storage_migration_service.rs
+486
View File
@@ -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<RwLock<MigrationState>>` + 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<PgPool>,
source: Arc<dyn BlobStorageBackend>,
storage_settings: Arc<StorageSettingsService>,
}
impl StorageMigrationService {
pub fn new(
pool: Arc<PgPool>,
source: Arc<dyn BlobStorageBackend>,
storage_settings: Arc<StorageSettingsService>,
) -> 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<Self>,
registry: &JobRegistry,
provider: &Arc<dyn JobStoreProvider>,
) -> Arc<Self> {
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<u64> {
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<Vec<u8>>,
) -> RunOutcome {
// No-op guard — refuse when the effective (target) config
// points at the same physical storage as the source (boot
// config). Without this, a misclick on an S3 deployment
// issues one HEAD per blob for zero useful work — cheap on
// local, expensive on remote. Same-type-different-location
// migrations (local dir change, S3 bucket change) pass this
// check and proceed normally.
match self.storage_settings.is_source_target_identical().await {
Ok(true) => {
tracing::warn!(
target: "audit",
event = "storage_migration.refused_noop",
run_id = %store.run_id(),
"storage_migration refused: source and target point at the same storage"
);
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<String> = 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(())
}
src/interfaces/api/handlers/admin_handler.rs
+319 -185
View File
@@ -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<Arc<AppState>> {
.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<Arc<AppState>>,
) -> Result<impl IntoResponse, AppError> {
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<String> = 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<Arc<AppState>>,
Json(dto): Json<StartMigrationDto>,
Json(_dto): Json<StartMigrationDto>,
) -> Result<impl IntoResponse, AppError> {
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<Arc<AppState>>,
) -> Result<impl IntoResponse, AppError> {
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<Arc<AppState>>,
) -> Result<impl IntoResponse, AppError> {
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<Arc<AppState>>,
) -> Result<impl IntoResponse, AppError> {
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<AppState>,
) -> Result<axum::response::Response, AppError> {
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<MigrationVerifyResult, crate::common::errors::DomainError> {
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<String>,
pub size_mismatches: Vec<String>,
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<String>,
) -> 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<impl IntoResponse, AppError> {
})))
}
/// Helper: build a BlobStorageBackend from StorageConfig.
fn build_backend_from_config(
config: &crate::common::config::StorageConfig,
) -> Result<
std::sync::Arc<dyn crate::application::ports::blob_storage_ports::BlobStorageBackend>,
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}`.
///
src/interfaces/api/mod.rs
-1
View File
@@ -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,