From bc481bdd990799ce067841dc406577d9e2cef750 Mon Sep 17 00:00:00 2001 From: Edouard Vanbelle Date: Sun, 2 Aug 2026 00:45:52 +0200 Subject: [PATCH] feat(job): fix key rotation on local storage (replace blob) --- src/application/ports/blob_storage_ports.rs | 36 ++++++++ .../services/encrypted_blob_backend.rs | 20 +++++ .../services/local_blob_backend.rs | 90 +++++++++++++++++++ .../services/storage_rotate_service.rs | 20 +++-- 4 files changed, 160 insertions(+), 6 deletions(-) diff --git a/src/application/ports/blob_storage_ports.rs b/src/application/ports/blob_storage_ports.rs index 5e2b9ea7..16f7df1d 100644 --- a/src/application/ports/blob_storage_ports.rs +++ b/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> { + 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`: diff --git a/src/infrastructure/services/encrypted_blob_backend.rs b/src/infrastructure/services/encrypted_blob_backend.rs index 6f7ae38c..6643f50a 100644 --- a/src/infrastructure/services/encrypted_blob_backend.rs +++ b/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> + 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], diff --git a/src/infrastructure/services/local_blob_backend.rs b/src/infrastructure/services/local_blob_backend.rs index af79cd2f..6109757b 100644 --- a/src/infrastructure/services/local_blob_backend.rs +++ b/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> + 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], diff --git a/src/infrastructure/services/storage_rotate_service.rs b/src/infrastructure/services/storage_rotate_service.rs index bf66ad7b..5f0e02b0 100644 --- a/src/infrastructure/services/storage_rotate_service.rs +++ b/src/infrastructure/services/storage_rotate_service.rs @@ -376,13 +376,21 @@ impl RecoverableJobHandler for StorageRotateService { continue; } - // Rewrite via the standard write path — atomic - // replace at the same object key. `put_blob_from_bytes` - // frames the plaintext with the head pair's format - // (encrypted-v1 or plaintext-v1) and hands the - // resulting bytes to the inner backend. + // Rewrite via the atomic-replace write path. + // + // **NOT** `put_blob_from_bytes`: that variant is + // idempotent-skip (`O_CREAT|O_EXCL` on + // `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;