feat(job): fix key rotation on local storage (replace blob)

src/application/ports/blob_storage_ports.rs

@@ -128,6 +128,42 @@ pub trait BlobStorageBackend: Send + Sync + 'static {
         self.put_blob_from_bytes(hash, data)
     }
 
+    /// Store a blob from in-memory bytes, **replacing** any existing
+    /// object at that hash. Distinct from [`Self::put_blob_from_bytes`]:
+    /// the standard variant is idempotent-skip (correct for uploads —
+    /// same plaintext always produces bytes that decrypt back to the
+    /// same plaintext), whereas this variant is required by callers
+    /// that need the on-disk BYTES to change even when the CONTENT
+    /// hash doesn't:
+    ///
+    /// * `storage_rotate` — rewrites every blob under the head pair's
+    ///   format (legacy → v1 header, old key → new key, plaintext ↔
+    ///   encrypted). If the target's `put_blob_from_bytes` silently
+    ///   skipped, rotation would report success while leaving the old
+    ///   format on disk.
+    /// * `storage_migration` — same story when a target already has a
+    ///   blob at that hash from an earlier state (Ed hit this on
+    ///   2026-08-01 in the S3 → local migration test).
+    ///
+    /// Must be **atomic** — a concurrent reader must see either the
+    /// old bytes or the new bytes, never a truncated partial write.
+    /// On POSIX that's a `write-to-tempfile + rename(2)` pattern; on
+    /// object storage (S3, Azure) it's a straight `PUT` (already
+    /// overwrites atomically).
+    ///
+    /// Default: delegates to `put_blob_from_bytes`. That default is
+    /// CORRECT for backends whose `put_blob_from_bytes` already
+    /// overwrites (S3, Azure — object stores overwrite on PUT by
+    /// default). Backends whose `put_blob_from_bytes` is
+    /// idempotent-skip (like `LocalBlobBackend`) MUST override.
+    fn put_blob_from_bytes_replace(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> BoxFut<'_, Result<u64, DomainError>> {
+        self.put_blob_from_bytes(hash, data)
+    }
+
     /// Make previously written blobs durable in one batched operation.
     ///
     /// Durability barrier for blobs written via `put_blob_from_bytes_unsynced`:

src/infrastructure/services/encrypted_blob_backend.rs

@@ -556,6 +556,26 @@ impl BlobStorageBackend for EncryptedBlobBackend {
         })
     }
 
+    /// Frame the plaintext with the head pair's format (encrypted-v1
+    /// or plaintext-v1), then delegate the atomic replace to the
+    /// inner backend. Used by `storage_rotate` to actually change the
+    /// on-disk bytes — `put_blob_from_bytes` would silently no-op on
+    /// `LocalBlobBackend` when the object key already exists.
+    fn put_blob_from_bytes_replace(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let inner = self.inner.clone();
+        let hash = hash.to_string();
+        let head_cipher = self.head_cipher.clone();
+        let head_key_fp = self.head_key_fp;
+        Box::pin(async move {
+            let out = frame_write(head_cipher, head_key_fp, data.to_vec()).await?;
+            inner.put_blob_from_bytes_replace(&hash, out).await
+        })
+    }
+
     fn sync_blobs(
         &self,
         hashes: &[String],

src/infrastructure/services/local_blob_backend.rs

@@ -422,6 +422,96 @@ impl BlobStorageBackend for LocalBlobBackend {
         })
     }
 
+    /// **Atomic replace**: write to a same-directory tempfile, fsync,
+    /// then `rename(2)` over the target. `write_blob_bytes`'s
+    /// `O_CREAT|O_EXCL` idempotent-skip (the right choice for uploads)
+    /// silently no-ops when the target already exists — wrong for
+    /// callers like `storage_rotate` that need the bytes to change.
+    /// See the trait doc for the full picture.
+    ///
+    /// Tempfile lives beside the target under the same shard directory
+    /// so `rename` is a cheap same-filesystem operation (never an
+    /// EXDEV cross-device copy fallback). The tempfile name embeds
+    /// the process pid + a monotonic counter so parallel replaces on
+    /// the same hash from different tasks don't clobber each other.
+    fn put_blob_from_bytes_replace(
+        &self,
+        hash: &str,
+        data: Bytes,
+    ) -> Pin<Box<dyn std::future::Future<Output = Result<u64, DomainError>> + Send + '_>> {
+        let hash = hash.to_owned();
+        Box::pin(async move {
+            let blob_path = self.blob_path(&hash);
+            let size = data.len() as u64;
+
+            // Tempfile in the SAME directory as the target → rename is
+            // cheap same-filesystem, never EXDEV. Counter ensures
+            // uniqueness under parallel replaces (rare — rotate is
+            // sequential per-blob today, but future concurrency won't
+            // corrupt).
+            static REPLACE_COUNTER: std::sync::atomic::AtomicU64 =
+                std::sync::atomic::AtomicU64::new(0);
+            let counter = REPLACE_COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+            let tmp_path = blob_path.with_file_name(format!(
+                "{}.replace.{}.{}.tmp",
+                hash,
+                std::process::id(),
+                counter
+            ));
+
+            // Create + write + fsync the tempfile. `create_new(true)`
+            // stays here to catch the astronomically-unlikely case of
+            // two tasks colliding on the same counter value (belt-and-
+            // braces; the pid+counter naming already prevents it).
+            {
+                let mut tmp = fs::File::options()
+                    .write(true)
+                    .create_new(true)
+                    .open(&tmp_path)
+                    .await
+                    .map_err(|e| {
+                        DomainError::internal_error(
+                            "Blob",
+                            format!("Failed to create replace-tmp: {}", e),
+                        )
+                    })?;
+                if let Err(e) = tmp.write_all(&data).await {
+                    let _ = fs::remove_file(&tmp_path).await;
+                    return Err(DomainError::internal_error(
+                        "Blob",
+                        format!("Failed to write replace-tmp: {}", e),
+                    ));
+                }
+                if let Err(e) = tmp.sync_all().await {
+                    let _ = fs::remove_file(&tmp_path).await;
+                    return Err(DomainError::internal_error(
+                        "Blob",
+                        format!("Failed to fsync replace-tmp: {}", e),
+                    ));
+                }
+            }
+
+            // Atomic replace. On POSIX `rename(2)` is atomic within a
+            // filesystem — a concurrent reader sees either the old or
+            // new bytes, never a truncated view. Older bytes drop out
+            // as soon as no reader holds an open fd.
+            if let Err(e) = fs::rename(&tmp_path, &blob_path).await {
+                let _ = fs::remove_file(&tmp_path).await;
+                return Err(DomainError::internal_error(
+                    "Blob",
+                    format!("Failed to atomically replace blob: {}", e),
+                ));
+            }
+
+            // fsync the parent directory so the dirent change (i.e. the
+            // rename result) survives a power loss, same discipline as
+            // the create path in `put_blob_from_bytes`.
+            fsync_parent_dir(&blob_path).await;
+
+            Ok(size)
+        })
+    }
+
     fn sync_blobs(
         &self,
         hashes: &[String],

src/infrastructure/services/storage_rotate_service.rs

@@ -376,13 +376,21 @@ impl RecoverableJobHandler for StorageRotateService {
                     continue;
                 }
 
-                // Rewrite via the standard write path — atomic
+                // Rewrite via the atomic-replace write path.
-                // replace at the same object key. `put_blob_from_bytes`
+                //
-                // frames the plaintext with the head pair's format
+                // **NOT** `put_blob_from_bytes`: that variant is
-                // (encrypted-v1 or plaintext-v1) and hands the
+                // idempotent-skip (`O_CREAT|O_EXCL` on
-                // resulting bytes to the inner backend.
+                // `LocalBlobBackend`) — correct for uploads (same
+                // plaintext ↔ any ciphertext at hash decrypts back)
+                // but a silent no-op for us. Rotate NEEDS the on-disk
+                // bytes to change (legacy → v1 header, old key → new
+                // key, plaintext ↔ encrypted). Ed hit this on
+                // 2026-08-02: rotation reported success in 9s but
+                // every blob on disk still had the legacy shape.
+                // `put_blob_from_bytes_replace` writes to a tempfile
+                // + atomic `rename(2)`s over the existing object key.
                 if let Err(e) = wrapper
-                    .put_blob_from_bytes(hash, Bytes::from(plaintext.to_vec()))
+                    .put_blob_from_bytes_replace(hash, Bytes::from(plaintext.to_vec()))
                     .await
                 {
                     failed_count += 1;