feat(storage): hot swap backend on migrate

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;

src/infrastructure/services/storage_migration_service.rs

@@ -84,14 +84,19 @@ const BATCH_SIZE: i64 = 100;
 
 pub struct StorageMigrationService {
     pool: Arc<PgPool>,
-    /// 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<dyn>`. 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<dyn BlobStorageBackend>,
-    /// 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<RwLock<String>>`
+    /// 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<std::sync::RwLock<String>>,
     /// 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<AtomicBool>,
+    /// 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<dyn ...>`
+    /// 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<PgPool>,
         source: Arc<dyn BlobStorageBackend>,
-        active_backend_name: String,
+        active_backend_name: Arc<std::sync::RwLock<String>>,
         storage_entries: Vec<NamedStorageEntry>,
         storage_path_fallback: PathBuf,
         migration_readonly: Arc<AtomicBool>,
+        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<dyn BlobStorageBackend>,
         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
     }

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<dyn BlobStorageBackend>` pointing at a specific concrete
+//! backend; it holds `Arc<SwappableBlobBackend>`, 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<T>` was the first choice here — it's
+/// lock-free — but it requires `T: Sized`, which `dyn ...` is not.
+/// The `Arc<dyn Trait>` 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<Arc<dyn BlobStorageBackend>>,
+}
+
+impl SwappableBlobBackend {
+    /// Wrap an initial backend. Every call goes to this one until
+    /// [`Self::swap`] replaces it.
+    pub fn new(initial: Arc<dyn BlobStorageBackend>) -> 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<dyn BlobStorageBackend>) {
+        // 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<dyn BlobStorageBackend> {
+        self.inner
+            .read()
+            .unwrap_or_else(std::sync::PoisonError::into_inner)
+            .clone()
+    }
+}
+
+/// Consumers of `Arc<dyn BlobStorageBackend>` 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<u64, DomainError>> {
+        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<u64, DomainError>> {
+        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<u64, DomainError>> {
+        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<BlobStream, DomainError>> {
+        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<u64>,
+    ) -> BoxFut<'_, Result<BlobStream, DomainError>> {
+        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<bool, DomainError>> {
+        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<u64, DomainError>> {
+        let inner = self.current();
+        let hash = hash.to_owned();
+        Box::pin(async move { inner.blob_size(&hash).await })
+    }
+
+    fn health_check(&self) -> BoxFut<'_, Result<StorageHealthStatus, DomainError>> {
+        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<PathBuf> {
+        self.current().local_blob_path(hash)
+    }
+
+    fn read_prefetch(&self) -> usize {
+        self.current().read_prefetch()
+    }
+
+    fn list_blob_hashes(
+        &self,
+        cursor: Option<String>,
+        limit: usize,
+    ) -> BoxFut<'_, Result<BlobListPage, DomainError>> {
+        let inner = self.current();
+        Box::pin(async move { inner.list_blob_hashes(cursor, limit).await })
+    }
+}
+