diff --git a/frontend/src/routes/admin/[[tab]]/+page.svelte b/frontend/src/routes/admin/[[tab]]/+page.svelte index f6e2d890..e55196ce 100644 --- a/frontend/src/routes/admin/[[tab]]/+page.svelte +++ b/frontend/src/routes/admin/[[tab]]/+page.svelte @@ -430,7 +430,7 @@ t( 'admin.storage_migrate_confirm', { name }, - 'Migrate all blobs to `{{name}}` and set it as the active entry? The server enters read-only mode during the copy; restart is required to finish cutover.' + 'Migrate all blobs to `{{name}}` and set it as the active entry? The server enters read-only mode during the copy; the live backend swaps automatically on completion (no restart needed).' ) ) ) @@ -1891,6 +1891,12 @@ ══════════════════════════════════════════════════════════ -->

{t('admin.storage_tab', 'Storage entries')}

+

+ {t( + 'admin.storage_move_hint', + 'To move to another backend storage: declare a new entry in your `.env` (keep the current one), restart the server so it picks it up, then trigger a migration from this page. Cutover happens automatically when the copy completes — no second restart needed.' + )} +

{#if !storage}

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

{:else if !storage.entries || storage.entries.length === 0} @@ -2136,27 +2142,12 @@ {/if}
- {#if migration?.status === 'completed' && storage.migration_readonly} - -
-

- - {t('admin.mig_cutover_done_title', 'Migration complete — restart to switch')} -

-

- {t( - 'admin.mig_cutover_done_body', - { active: storage.active_entry_name ?? '?' }, - 'The DB pointer now names `{{active}}` as the active backend, but the running process is still bound to the previous entry. Restart the server to complete the cutover; boot picks up the new active entry and clears read-only mode automatically.' - )} -

-
- {/if} + {/if} {#if storageMsg}

{storageMsg.text}

diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index 29fa87fa..5e2b9ea7 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/src/application/ports/blob_storage_ports.rs @@ -67,7 +67,7 @@ pub struct BlobListPage { } /// Boxed future alias used by [`BlobStorageBackend`] to keep the trait dyn-compatible. -type BoxFut<'a, T> = Pin + Send + 'a>>; +pub type BoxFut<'a, T> = Pin + Send + 'a>>; /// Pinned boxed byte stream — the return type for blob reads. pub type BlobStream = Pin> + Send>>; diff --git a/src/application/services/storage_settings_service.rs b/src/application/services/storage_settings_service.rs index 4c13978b..a96af68e 100644 --- a/src/application/services/storage_settings_service.rs +++ b/src/application/services/storage_settings_service.rs @@ -32,10 +32,11 @@ pub struct StorageSettingsService { /// restart, per `docs/plan/storage-multi-entry.md`). Empty when /// running in the pre-multi-entry legacy path. storage_entries: Vec, - /// Name of the entry the LIVE backend is bound to (matches - /// `CoreServices.active_backend_name`). Empty string / "legacy" - /// for the zero-entries path. - active_entry_name: String, + /// Shared handle to `CoreServices.active_backend_name`. Reads + /// snapshot the current value on each admin GET so a hot-swap + /// cutover is immediately visible in the UI without waiting for + /// a refresh. Empty string / "legacy" for the zero-entries path. + active_entry_name: Arc>, /// Shared readonly flag — read into the admin DTO so the UI can /// render a "server in migration read-only mode" banner. Same /// atomic as `AppState.migration_readonly`; changes made by the @@ -50,7 +51,7 @@ impl StorageSettingsService { env_storage_config: StorageConfig, dedup_service: Arc, storage_entries: Vec, - active_entry_name: String, + active_entry_name: Arc>, migration_readonly: Arc, ) -> Self { Self { @@ -253,6 +254,14 @@ impl StorageSettingsService { // exactly one entry when we're in multi-entry mode; matches // nothing when running the zero-entries legacy path, which // is expected — the frontend hides the entries table then). + // Snapshot the active name once; every entry below uses it + // for the is_active flag AND we echo it on the top-level + // DTO. RwLock is held only for the clone. + let active_entry_name = self + .active_entry_name + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); let entries: Vec = self .storage_entries .iter() @@ -263,7 +272,7 @@ impl StorageSettingsService { StorageBackendType::S3 => "s3".to_string(), StorageBackendType::Azure => "azure".to_string(), }, - is_active: e.name == self.active_entry_name, + is_active: e.name == active_entry_name, encryption_enabled: e.encryption_key_base64.is_some(), location_hint: entry_location_hint(e), }) @@ -275,7 +284,7 @@ impl StorageSettingsService { total_bytes_stored: stats.total_bytes_stored, dedup_ratio: stats.dedup_ratio, entries, - active_entry_name: self.active_entry_name.clone(), + active_entry_name, migration_readonly: self.migration_readonly.load(Ordering::Relaxed), }) } diff --git a/src/common/di.rs b/src/common/di.rs index 459537a2..acc79189 100644 --- a/src/common/di.rs +++ b/src/common/di.rs @@ -323,6 +323,11 @@ impl AppServiceFactory { active_backend_name = entry.name.clone(); build_entry_backend(entry, &self.storage_path) }; + // Shared mutable handle to the active-entry name. Wrapped + // here rather than at the field type so the two String + // literal writes above stay simple; wrapping happens once + // just before the struct init. + let active_backend_name = Arc::new(std::sync::RwLock::new(active_backend_name)); // Stack decorators: retry → encryption → cache (inner-to-outer). // @@ -407,6 +412,21 @@ impl AppServiceFactory { tracing::info!("Blob storage LRU disk cache enabled"); } + // Wrap the fully-decorated stack in the hot-swap wrapper. + // Every downstream consumer holds `Arc` + // as before; the wrapper is transparent from their point of + // view. The `Arc` reference we retain + // here (stored on `AppState.blob_backend_hot_swap`) is what + // the migration handler calls `.swap()` on when cutover + // completes — no restart needed. See + // `swappable_blob_backend.rs` for the delegation contract. + let blob_backend_hot_swap = Arc::new( + crate::infrastructure::services::swappable_blob_backend::SwappableBlobBackend::new( + blob_backend, + ), + ); + let blob_backend: Arc = blob_backend_hot_swap.clone(); + // Blob lifecycle — thumbnail disk-file cleanup when blob ref_count hits zero. // ThumbnailService (not ThumbnailRefreshHook) is used here to avoid a circular // Arc: DedupService→BlobLifecycleService→ThumbnailRefreshHook→DedupService. @@ -558,6 +578,7 @@ impl AppServiceFactory { job_registry, job_store_provider, blob_backend: blob_backend_for_consistency, + blob_backend_hot_swap, active_backend_name, }) } @@ -2232,6 +2253,7 @@ impl AppServiceFactory { app_state.core.config.storage_entries.clone(), self.storage_path.clone(), app_state.migration_readonly.clone(), + app_state.core.blob_backend_hot_swap.clone(), ), ) .register_recoverable_job(&app_state.core.job_registry, &job_store_provider_dyn) @@ -2483,12 +2505,18 @@ impl AppServiceFactory { // active_backend_name. Absence (Unset) is treated as "no // mismatch to complain about" — the boot fallback already // picked the first entry. + let booted_active = app_state + .core + .active_backend_name + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); let db_active_matches = { use crate::infrastructure::services::entry_backend::{ ActiveEntry, resolve_active_entry, }; match resolve_active_entry(&pool, &app_state.core.config.storage_entries).await { - Ok(ActiveEntry::Explicit(e)) => e.name == app_state.core.active_backend_name, + Ok(ActiveEntry::Explicit(e)) => e.name == booted_active, Ok(ActiveEntry::Unset) => true, Err(_) => false, } @@ -2504,7 +2532,7 @@ impl AppServiceFactory { tracing::info!( target: "audit", event = "storage.migration_readonly.cleared_at_boot", - active = %app_state.core.active_backend_name, + active = %booted_active, "🧊 migration_readonly cleared at boot: no in-flight migration + \ active_backend_name matches booted entry (cutover complete on prior boot)" ); @@ -2581,7 +2609,22 @@ pub struct CoreServices { /// `create_core_services` — notably `blobs_consistency` in /// `build_app_state` — can probe `blob_exists()` / re-hash bytes /// through the same stack DedupService uses. + /// + /// Concretely this is the hot-swap wrapper coerced to + /// `Arc`; a migration cutover replaces the inner + /// backend via [`Self::blob_backend_hot_swap`] and every future + /// call through this `blob_backend` sees the new inner. pub blob_backend: Arc, + /// Typed handle to the hot-swap wrapper. Distinct from + /// [`Self::blob_backend`] only in its declared type: the raw + /// wrapper struct instead of `dyn BlobStorageBackend`. Same + /// underlying instance, so a call to `.swap(new)` here is + /// immediately visible through the trait-object handle above. + /// The migration handler is the only intended caller — it flips + /// the pointer on `RunOutcome::Completed`, so restart is no + /// longer required for cutover. + pub blob_backend_hot_swap: + Arc, /// Name of the storage entry the LIVE `blob_backend` was built /// from. Populated at boot: either from /// `admin_settings.storage.active_backend_name` when set, or the @@ -2591,7 +2634,14 @@ pub struct CoreServices { /// config at all). Migration handler consumes this to enforce the /// "target != active" no-op guard by name; without needing to /// re-read DB on every trigger. - pub active_backend_name: String, + /// + /// Wrapped in `Arc>` so the migration handler can + /// update it on hot-swap — subsequent name-based guards (a + /// second migration triggered by the admin after the first cut + /// over) see the new active without a restart. Read pattern: + /// acquire the read lock, clone the inner String, release the + /// lock, use the clone across await points. + pub active_backend_name: Arc>, } /// Container for repository services diff --git a/src/infrastructure/services/mod.rs b/src/infrastructure/services/mod.rs index a9a782d6..924c3115 100644 --- a/src/infrastructure/services/mod.rs +++ b/src/infrastructure/services/mod.rs @@ -11,6 +11,7 @@ pub mod dedup_service; pub mod drives_consistency_service; pub mod encrypted_blob_backend; pub mod entry_backend; +pub mod swappable_blob_backend; pub mod exif_service; pub mod face_geometry; pub mod face_indexing_service; diff --git a/src/infrastructure/services/storage_migration_service.rs b/src/infrastructure/services/storage_migration_service.rs index e0be47ab..a8cfac4d 100644 --- a/src/infrastructure/services/storage_migration_service.rs +++ b/src/infrastructure/services/storage_migration_service.rs @@ -84,14 +84,19 @@ const BATCH_SIZE: i64 = 100; pub struct StorageMigrationService { pool: Arc, - /// Backend the running app is bound to — the migration COPIES - /// FROM this. Set once at boot and never changes for the - /// process's lifetime (cutover requires a restart, per plan). + /// Backend the running app is bound to at handler-construction + /// time. Refers to the hot-swap wrapper when multi-entry is + /// active, so this read stays live across cutovers even if + /// stored as `Arc`. Only used to identify the source + /// entry's `backend_type()` for audit lines; the actual copy + /// path reads from `self.pool` and writes to the target + /// backend built via `build_entry_backend`. source: Arc, - /// Name of the currently-active entry (i.e. the one `source` - /// corresponds to). Used to refuse a same-name target at run - /// start. Same reasoning as `source` — locked at boot. - active_backend_name: String, + /// Name of the currently-active entry. Shared `Arc>` + /// with `CoreServices.active_backend_name` — a hot-swap-mutation + /// on cutover is visible here without reconstructing the handler. + /// Read via `.read().clone()` at the top of each run. + active_backend_name: Arc>, /// All entries declared in env, held as a snapshot for name /// lookup during migration. Immutable per-deploy — matches /// `AppConfig.storage_entries`. @@ -104,12 +109,19 @@ pub struct StorageMigrationService { /// Shared `AppState.migration_readonly` handle. Handler flips /// this atomic (and persists to DB) at run start once all /// guards pass, so writes across the whole app get refused by - /// the AuthZ short-circuit for the duration of the copy. Kept - /// ON when Completed — the boot-clear rule (slice 4) resets it - /// on the next restart after cutover, so operators can't - /// accidentally re-enable writes on the OLD backend while the - /// pointer already says the NEW one is active. + /// the AuthZ short-circuit for the duration of the copy. On + /// `RunOutcome::Completed`, the handler hot-swaps the runtime + /// backend + clears this flag in one step — no restart. migration_readonly: Arc, + /// Typed handle to the runtime blob-backend wrapper. The + /// migration handler calls `.swap()` on this at cutover so + /// subsequent user writes go to the target entry without a + /// restart. Shared with `CoreServices.blob_backend_hot_swap` — + /// same instance the coerced `blob_backend: Arc` + /// delegates through. + blob_backend_hot_swap: Arc< + crate::infrastructure::services::swappable_blob_backend::SwappableBlobBackend, + >, } impl StorageMigrationService { @@ -117,10 +129,13 @@ impl StorageMigrationService { pub fn new( pool: Arc, source: Arc, - active_backend_name: String, + active_backend_name: Arc>, storage_entries: Vec, storage_path_fallback: PathBuf, migration_readonly: Arc, + blob_backend_hot_swap: Arc< + crate::infrastructure::services::swappable_blob_backend::SwappableBlobBackend, + >, ) -> Self { Self { pool, @@ -129,6 +144,7 @@ impl StorageMigrationService { storage_entries, storage_path_fallback, migration_readonly, + blob_backend_hot_swap, } } @@ -229,19 +245,30 @@ impl RecoverableJobHandler for StorageMigrationService { } }; + // Snapshot the current active name for the rest of this + // run. The lock is held only for the clone; every subsequent + // reference reads from this local. A hot-swap that fires + // mid-run (e.g., a second migration starting after this one + // completes) doesn't reshape our decisions from underneath. + let active_backend_name = self + .active_backend_name + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); + // First-line guard: target name equals the currently-active // entry. Silent no-op if we let it through — the app would // walk every blob and skip because `target.blob_exists` is // trivially true (target = live source). Even on the same // local disk that's a lot of syscalls for no reason; on S3 // it costs one HEAD per blob for zero copies. - if target_name == self.active_backend_name { + if target_name == active_backend_name { tracing::warn!( target: "audit", event = "storage_migration.refused_noop", run_id = %store.run_id(), target_name = %target_name, - active = %self.active_backend_name, + active = %active_backend_name, "storage_migration refused: target equals the currently-active entry" ); return RunOutcome::Failed { @@ -277,7 +304,7 @@ impl RecoverableJobHandler for StorageMigrationService { let source_entry = self .storage_entries .iter() - .find(|e| e.name == self.active_backend_name); + .find(|e| e.name == active_backend_name); // Second-line guard: physical-identity check for the // encryption-differs case. Two entries with different names @@ -305,15 +332,14 @@ impl RecoverableJobHandler for StorageMigrationService { event = "storage_migration.refused_same_physical_storage", run_id = %store.run_id(), target_name = %target_name, - source_name = %self.active_backend_name, + source_name = %active_backend_name, encryption_differs = key_differs, "storage_migration refused: named target differs from source but physical storage matches" ); return RunOutcome::Failed { message: format!( "target entry `{target_name}` names a different entry than the active \ - `{}`, but they point at the same physical storage{hint}.", - self.active_backend_name, + `{active_backend_name}`, but they point at the same physical storage{hint}." ), }; } @@ -356,7 +382,7 @@ impl RecoverableJobHandler for StorageMigrationService { run_id = %store.run_id(), target_name = %target_name, "🚧 migration_readonly engaged: writes across the whole app are refused until \ - cutover completes and the operator restarts" + cutover hot-swap completes" ); let source_kind = self.source.backend_type(); @@ -365,13 +391,12 @@ impl RecoverableJobHandler for StorageMigrationService { target: "audit", event = "storage_migration.run_started", run_id = %store.run_id(), - source_name = %self.active_backend_name, + source_name = %active_backend_name, target_name = %target_name, source_kind = source_kind, target_kind = target_kind, resuming = !is_fresh, - "storage_migration starting {} ({source_kind}) → {target_name} ({target_kind})", - self.active_backend_name, + "storage_migration starting {active_backend_name} ({source_kind}) → {target_name} ({target_kind})" ); // Cursor = the last-visited blob hash, UTF-8-encoded. On resume @@ -453,6 +478,8 @@ impl RecoverableJobHandler for StorageMigrationService { .finish_completed( store, &target_name, + &active_backend_name, + target.clone(), copied_count, skipped_count, failed_count, @@ -593,6 +620,8 @@ impl RecoverableJobHandler for StorageMigrationService { .finish_completed( store, &target_name, + &active_backend_name, + target.clone(), copied_count, skipped_count, failed_count, @@ -606,30 +635,45 @@ impl RecoverableJobHandler for StorageMigrationService { impl StorageMigrationService { /// Terminal successful path — reached from both Completed sites - /// in the batch loop (empty-first-batch and short-batch). Flips - /// the runtime `active_backend_name` pointer to the target entry - /// so the NEXT boot picks it up. Leaves `migration_readonly` ON - /// — the boot-clear rule (slice 4) drops it after the operator - /// restart when no in-flight run remains AND the DB pointer - /// matches the entry the app booted onto. + /// in the batch loop (empty-first-batch and short-batch). /// - /// Pointer-write failure is FATAL to the outcome. Reporting - /// `Completed` while the DB still says the old entry is active - /// would strand the migrated bytes: the next boot would come up - /// on the OLD backend (writes to old!), while the operator - /// thinks cutover is done. `Failed` keeps the situation legible: - /// admin sees the error, can retry the pointer write, then - /// restart. + /// Four things happen here, in order, and each has a fail + /// posture: + /// + /// 1. Persist `active_backend_name = target_name` to + /// `admin_settings`. Fatal on error — reporting `Completed` + /// while the DB still says the old entry is active would + /// strand the migrated bytes (next boot would come up on the + /// OLD backend). Operator retries — the walk short-circuits + /// on already-present blobs, so the retry is cheap. + /// 2. **Hot-swap** the runtime blob backend to the target. The + /// already-initialized `target` backend is passed in from + /// `run_resumable` (built via `build_entry_backend`, so + /// encryption + config are set up); `SwappableBlobBackend`'s + /// `swap` is a `RwLock::write` — instantaneous. In-flight + /// reads holding the old inner `Arc` finish against the old + /// backend; new operations see the new one. + /// 3. Update `active_backend_name` in the shared `RwLock` so a + /// second migration triggered right after (target != active) + /// sees the new active name without a restart. + /// 4. Persist + clear `migration_readonly` — writes resume, + /// against the new backend. If DB persist fails at this step, + /// log a warning and clear the in-memory flag anyway: the + /// next boot's clear rule will fix the DB row on restart if + /// it's still stale. #[allow(clippy::too_many_arguments)] async fn finish_completed( &self, store: &dyn JobStore, target_name: &str, + previous_active: &str, + target_backend: Arc, copied: u64, skipped: u64, failed: u64, source_missing: u64, ) -> RunOutcome { + // 1. DB pointer. if let Err(e) = persist_active_backend_name(self.pool.as_ref(), target_name).await { return RunOutcome::Failed { message: format!( @@ -640,18 +684,49 @@ impl StorageMigrationService { ), }; } + + // 2. Runtime hot-swap. + self.blob_backend_hot_swap.swap(target_backend); + + // 3. In-memory active-name mirror. + { + let mut guard = self + .active_backend_name + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *guard = target_name.to_string(); + } + + // 4. Drop read-only. In this order (after swap) so no write + // slips through against the OLD backend between "readonly + // off" and "backend swapped". + let readonly_persisted = + persist_migration_readonly(self.pool.as_ref(), false).await.is_ok(); + self.migration_readonly.store(false, Ordering::Relaxed); + + if !readonly_persisted { + tracing::warn!( + target: "oxicloud::migration", + event = "storage_migration.readonly_clear_persist_failed", + run_id = %store.run_id(), + "cleared migration_readonly in memory (writes allowed) but the DB persist \ + failed. If the server crashes before next boot, boot will re-seed the flag \ + to true; boot-clear rule then flips it since active-matches and no in-flight." + ); + } + tracing::info!( target: "audit", event = "storage_migration.completed", run_id = %store.run_id(), active_backend_name = target_name, - previous_active = %self.active_backend_name, + previous_active = previous_active, copied = copied, skipped = skipped, failed = failed, source_missing = source_missing, - "✅ storage_migration completed — active_backend_name = `{target_name}`. Restart the \ - server to switch the live backend (migration_readonly stays ON until then)." + "✅ storage_migration completed — hot-swapped runtime backend to `{target_name}`, \ + writes resumed. No restart required." ); RunOutcome::Completed } diff --git a/src/infrastructure/services/swappable_blob_backend.rs b/src/infrastructure/services/swappable_blob_backend.rs new file mode 100644 index 00000000..40c30d2c --- /dev/null +++ b/src/infrastructure/services/swappable_blob_backend.rs @@ -0,0 +1,225 @@ +//! `SwappableBlobBackend` — atomic hot-swap wrapper for `BlobStorageBackend`. +//! +//! Enables in-process cutover after a migration completes, without +//! restarting the server. The runtime never holds a raw +//! `Arc` pointing at a specific concrete +//! backend; it holds `Arc`, and every method +//! call snapshots the CURRENT inner backend from an `ArcSwap` before +//! delegating. +//! +//! Design contract (see `docs/plan/storage-multi-entry.md` +//! §"Read-only mode" cross-ref — the hot-swap upgrade removes the +//! restart step): +//! +//! * **Reads are lock-free.** `ArcSwap::load_full` bumps the strong +//! count on the current inner Arc and returns it — no lock, no +//! contention with concurrent readers, no contention with a swap. +//! * **In-flight operations complete on the OLD backend.** Each +//! method call clones the current inner Arc at the top; the +//! spawned future holds that clone for its whole lifetime. A swap +//! that happens mid-future doesn't affect that future — a `PUT` +//! that started on local completes on local, even after cutover +//! flipped the pointer to S3. +//! * **New operations after a swap see the NEW backend.** The +//! `ArcSwap::store` on `swap()` is atomic relative to +//! `ArcSwap::load_full` — no race window where a request sees a +//! torn state. +//! * **The old backend is dropped when the last in-flight future +//! holding it finishes.** Standard `Arc` refcount semantics. +//! +//! The wrapper delegates every method the `BlobStorageBackend` trait +//! declares. Two mildly interesting cases: +//! - `local_blob_path` — meaningful only for a local backend. If the +//! current inner is S3, returns `None`, same as any remote backend. +//! After a Local→S3 hot-swap, callers stop getting fast-paths; +//! they get streaming (correct fallback in every caller). +//! - `list_blob_hashes` — takes an opaque cursor whose format is +//! backend-specific. Swapping mid-enumeration would invalidate the +//! cursor. Not currently an issue because enumeration only runs +//! inside `backend_consistency` / `blobs_consistency`, which +//! snapshot at run start (they clone the Arc into their own local +//! `backend` for the whole loop, so a mid-run swap doesn't affect +//! them — same in-flight guarantee as any other operation). + +use std::path::{Path, PathBuf}; +use std::sync::{Arc, RwLock}; + +use bytes::Bytes; + +use crate::application::ports::blob_storage_ports::{ + BlobListPage, BlobStorageBackend, BlobStream, BoxFut, StorageHealthStatus, +}; +use crate::common::errors::DomainError; + +/// Atomic hot-swap wrapper around a concrete `BlobStorageBackend`. +/// +/// Reads are effectively lock-free — an uncontended `RwLock::read` +/// on a modern std impl is just a relaxed atomic bump; contention +/// only exists during the brief window of a swap. Swap is a bounded +/// `RwLock::write` that flips the inner `Arc` — no I/O, no waiting +/// for in-flight futures. In-flight futures already hold a clone of +/// the previous inner `Arc` (loaded when their method call started) +/// and complete against that; the old Arc drops naturally when the +/// last of them finishes. +/// +/// `arc-swap`'s `ArcSwap` was the first choice here — it's +/// lock-free — but it requires `T: Sized`, which `dyn ...` is not. +/// The `Arc` snapshot pattern via `RwLock` is the +/// standard workaround; performance is indistinguishable at the +/// per-request cadence of a blob backend. +pub struct SwappableBlobBackend { + inner: RwLock>, +} + +impl SwappableBlobBackend { + /// Wrap an initial backend. Every call goes to this one until + /// [`Self::swap`] replaces it. + pub fn new(initial: Arc) -> Self { + Self { + inner: RwLock::new(initial), + } + } + + /// Atomically replace the inner backend. Subsequent method calls + /// see `new`; futures already running against the previous inner + /// keep running to completion (the previous Arc is only fully + /// dropped when the last of them finishes). + /// + /// Callers responsible for `.initialize()` on `new` before the + /// swap — this method assumes the new backend is ready. + pub fn swap(&self, new: Arc) { + // A poisoned lock here would mean a panic during a previous + // swap — recover the guard to keep the swap surface robust + // (nobody should panic during a normal `store`, but a + // recovering-from-poison call still yields a working writer). + let mut guard = self + .inner + .write() + .unwrap_or_else(std::sync::PoisonError::into_inner); + *guard = new; + } + + /// Take a snapshot of the current inner backend. Cheap: one + /// `read()` + one strong-count bump on the returned Arc. Callers + /// can hold the resulting Arc across await points; the wrapper's + /// swap won't affect it. + pub fn current(&self) -> Arc { + self.inner + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone() + } +} + +/// Consumers of `Arc` get exactly today's +/// contract — the wrapper is transparent from their point of view. +impl BlobStorageBackend for SwappableBlobBackend { + fn initialize(&self) -> BoxFut<'_, Result<(), DomainError>> { + let inner = self.current(); + Box::pin(async move { inner.initialize().await }) + } + + fn put_blob(&self, hash: &str, source_path: &Path) -> BoxFut<'_, Result> { + let inner = self.current(); + let hash = hash.to_owned(); + let source_path = source_path.to_owned(); + Box::pin(async move { inner.put_blob(&hash, &source_path).await }) + } + + fn put_blob_from_bytes( + &self, + hash: &str, + data: Bytes, + ) -> BoxFut<'_, Result> { + let inner = self.current(); + let hash = hash.to_owned(); + Box::pin(async move { inner.put_blob_from_bytes(&hash, data).await }) + } + + fn put_blob_from_bytes_unsynced( + &self, + hash: &str, + data: Bytes, + ) -> BoxFut<'_, Result> { + let inner = self.current(); + let hash = hash.to_owned(); + Box::pin(async move { inner.put_blob_from_bytes_unsynced(&hash, data).await }) + } + + fn sync_blobs(&self, hashes: &[String]) -> BoxFut<'_, Result<(), DomainError>> { + let inner = self.current(); + let hashes = hashes.to_vec(); + Box::pin(async move { inner.sync_blobs(&hashes).await }) + } + + fn get_blob_stream(&self, hash: &str) -> BoxFut<'_, Result> { + let inner = self.current(); + let hash = hash.to_owned(); + Box::pin(async move { inner.get_blob_stream(&hash).await }) + } + + fn get_blob_range_stream( + &self, + hash: &str, + start: u64, + end: Option, + ) -> BoxFut<'_, Result> { + let inner = self.current(); + let hash = hash.to_owned(); + Box::pin(async move { inner.get_blob_range_stream(&hash, start, end).await }) + } + + fn delete_blob(&self, hash: &str) -> BoxFut<'_, Result<(), DomainError>> { + let inner = self.current(); + let hash = hash.to_owned(); + Box::pin(async move { inner.delete_blob(&hash).await }) + } + + fn blob_exists(&self, hash: &str) -> BoxFut<'_, Result> { + let inner = self.current(); + let hash = hash.to_owned(); + Box::pin(async move { inner.blob_exists(&hash).await }) + } + + fn blob_size(&self, hash: &str) -> BoxFut<'_, Result> { + let inner = self.current(); + let hash = hash.to_owned(); + Box::pin(async move { inner.blob_size(&hash).await }) + } + + fn health_check(&self) -> BoxFut<'_, Result> { + let inner = self.current(); + Box::pin(async move { inner.health_check().await }) + } + + fn backend_type(&self) -> &'static str { + // We return the CURRENT backend's static kind. Delegation + // means the returned string reflects the entry the swap + // pointer currently names. + // + // NB: `&'static str` is deliberate — every backend's kind is + // a compile-time constant ("local", "s3", "azure"), so this + // is safe. The trait signature makes it appear to outlive + // the `&self` borrow, but the value is a static so there's + // no lifetime hazard. + self.current().backend_type() + } + + fn local_blob_path(&self, hash: &str) -> Option { + self.current().local_blob_path(hash) + } + + fn read_prefetch(&self) -> usize { + self.current().read_prefetch() + } + + fn list_blob_hashes( + &self, + cursor: Option, + limit: usize, + ) -> BoxFut<'_, Result> { + let inner = self.current(); + Box::pin(async move { inner.list_blob_hashes(cursor, limit).await }) + } +} + diff --git a/src/interfaces/api/handlers/admin_handler.rs b/src/interfaces/api/handlers/admin_handler.rs index 2d1f91b1..c000e520 100644 --- a/src/interfaces/api/handlers/admin_handler.rs +++ b/src/interfaces/api/handlers/admin_handler.rs @@ -463,7 +463,14 @@ pub async fn start_migration( // failed run row. The handler's own checks are second-line // defence for the resume path where args aren't repeated. let entries = &state.core.config.storage_entries; - let active = &state.core.active_backend_name; + // Snapshot the active-name for the guard. Read lock held only + // for the clone. + let active = state + .core + .active_backend_name + .read() + .unwrap_or_else(std::sync::PoisonError::into_inner) + .clone(); if entries.iter().all(|e| e.name != dto.target_name) { let available = if entries.is_empty() { "(none)".to_string() @@ -479,7 +486,7 @@ pub async fn start_migration( dto.target_name ))); } - if dto.target_name == *active { + if dto.target_name == active { return Err(AppError::bad_request(format!( "target `{}` is the currently-active entry — pick a different entry to migrate to", dto.target_name