fix(kms): CAS Transit metadata writes and bound the metadata cache

Transit KV metadata writes were whole-record overwrites with no
precondition, so two nodes mutating the same key could silently clobber
each other's lifecycle state, and the process-local metadata cache had
neither a TTL nor a capacity bound, so a key disabled or scheduled for
deletion on one node stayed usable on every other node until restart.

- Replace write_metadata_to_kv with a versioned read
  (read_metadata_from_kv_versioned) plus a check-and-set write
  (cas_write_metadata_to_kv); mutate_key_metadata re-reads the
  authoritative record and re-runs the state gate on every attempt, and a
  lost CAS race retries with a fresh snapshot (bounded budget) instead of
  replaying the stale one.
- Migrate every read-modify-write caller: enable, disable, schedule and
  cancel deletion, rotate version bump, the expired-key tombstone, and
  both create paths (create-only CAS that read-confirms the winner on a
  lost race).
- Bound the metadata cache with moka (300s TTL, 1024 entries) and drop a
  key's entry when a transit data call reports it gone server-side.
- Fail closed when the synthesized-metadata fallback cannot be read or
  persisted: the fabricated Enabled record is only served after a durable
  create-only CAS write, closing the gate weakening documented as a KNOWN
  RISK; the persistence fallback for pre-metadata keys is kept.

Refs rustfs/backlog#1581 (part of rustfs/backlog#1562)

crates/kms/src/backends/local.rs

@@ -397,6 +397,20 @@ pub struct LocalKmsClient {
     /// Per-key write locks serializing read-modify-write updates within this
     /// process (see [`Self::lock_key_for_write`]).
     key_write_locks: Mutex<HashMap<String, Arc<tokio::sync::Mutex<()>>>>,
+    /// Directory-wide writer fence for backup export (see
+    /// [`Self::acquire_export_fence`]). Writers hold the read side; an export
+    /// snapshot holds the write side so it observes a single-generation view.
+    export_fence: Arc<tokio::sync::RwLock<()>>,
+}
+
+/// Guard pairing the export-fence read lock with a per-key write mutex.
+///
+/// Dropping it releases both, so every existing `lock_key_for_write` call
+/// site participates in the export fence without changes.
+#[must_use]
+struct KeyWriteGuard {
+    _fence: tokio::sync::OwnedRwLockReadGuard<()>,
+    _key: tokio::sync::OwnedMutexGuard<()>,
 }
 
 // pub(crate) so the backup contract tests can anchor the manifest's
@@ -463,6 +477,7 @@ impl LocalKmsClient {
             legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
             key_write_locks: Mutex::new(HashMap::new()),
+            export_fence: Arc::new(tokio::sync::RwLock::new(())),
         };
         client.validate_existing_keys().await?;
         Ok(client)
@@ -505,6 +520,7 @@ impl LocalKmsClient {
             legacy_master_cipher,
             dek_crypto: AesDekCrypto::new(),
             key_write_locks: Mutex::new(HashMap::new()),
+            export_fence: Arc::new(tokio::sync::RwLock::new(())),
         })
     }
 
@@ -515,12 +531,40 @@ impl LocalKmsClient {
     /// delete with a rewrite. Cross-process writers sharing a key directory
     /// remain unsupported. Entries live for the client's lifetime; the table
     /// is bounded by the number of distinct key ids this process touches.
-    async fn lock_key_for_write(&self, key_id: &str) -> tokio::sync::OwnedMutexGuard<()> {
+    async fn lock_key_for_write(&self, key_id: &str) -> KeyWriteGuard {
+        // Fence first, per-key mutex second: the ordering is uniform across
+        // all writers, so an export waiting on the write side can never
+        // deadlock with a writer holding a key mutex.
+        let fence = Arc::clone(&self.export_fence).read_owned().await;
         let lock = {
             let mut locks = self.key_write_locks.lock().expect("Local KMS key write lock table poisoned");
             Arc::clone(locks.entry(key_id.to_string()).or_default())
         };
-        lock.lock_owned().await
+        KeyWriteGuard {
+            _fence: fence,
+            _key: lock.lock_owned().await,
+        }
+    }
+
+    /// Block every key-directory writer while a backup export collects its
+    /// snapshot, so all records belong to one generation.
+    ///
+    /// Mutating operations hold the read side (via [`Self::lock_key_for_write`]
+    /// or [`Self::save_new_master_key`]); the export holds the write side only
+    /// for the collection phase, never while encrypting or writing the bundle.
+    pub(crate) async fn acquire_export_fence(&self) -> tokio::sync::OwnedRwLockWriteGuard<()> {
+        Arc::clone(&self.export_fence).write_owned().await
+    }
+
+    /// Key directory root, exposed for the backup export module.
+    pub(crate) fn key_directory(&self) -> &Path {
+        &self.config.key_dir
+    }
+
+    /// Absolute path of the master-key KDF salt file, exposed for the backup
+    /// export module.
+    pub(crate) fn master_key_salt_file(&self) -> PathBuf {
+        Self::master_key_salt_path(&self.config)
     }
 
     /// Derive a 256-bit key from the master key string using a persistent Argon2id salt.
@@ -797,6 +841,11 @@ impl LocalKmsClient {
     }
 
     async fn save_new_master_key(&self, master_key: &MasterKeyInfo, key_material: &[u8]) -> Result<()> {
+        // Creates never take the per-key write lock (`NoClobber` publishing
+        // already linearizes them), so they join the export fence here. This
+        // must stay the only fence acquisition on the create path: the fence
+        // read lock is not reentrant while an export waits for the write side.
+        let _fence = Arc::clone(&self.export_fence).read_owned().await;
         let key_path = self.master_key_path(&master_key.key_id)?;
         let content = self.encode_master_key(master_key, key_material)?;
         let temp_path = key_path.with_extension(format!("tmp-{}", uuid::Uuid::new_v4()));

crates/kms/src/backup/local_export.rs

@@ -0,0 +1,998 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Local backend backup export: sealed, KEK-protected bundle production.
+//!
+//! This is the producer side only; restore lives in a follow-up change. The
+//! admin API is not wired here either — callers construct the request and
+//! supply the backup KEK explicitly.
+//!
+//! # Bundle layout
+//!
+//! A bundle is a directory (simple to produce, artifacts stream one file at a
+//! time, and partial output is trivially recognizable because the manifest is
+//! written last):
+//!
+//! ```text
+//! <destination>/
+//!   manifest.json                      # sealed BackupManifest, written last
+//!   artifacts/keys/<key_id>.key.enc    # one per stored key record
+//!   artifacts/master-key.salt.enc      # present when the salt file exists
+//! ```
+//!
+//! # Artifact payload framing
+//!
+//! Every artifact payload is `nonce (12 bytes) || AES-256-GCM ciphertext`,
+//! encrypted under the caller-supplied backup KEK with an AAD binding of
+//! `(context, backup_id, snapshot_generation, artifact path)`, so an artifact
+//! cannot be swapped into another bundle or renamed within its own bundle.
+//! Records already encrypted at rest stay encrypted inside the wrap;
+//! plaintext-dev-only records become ciphertext-only in the bundle, which is
+//! their mandatory re-wrap under the backup KEK.
+//!
+//! # Write protocol
+//!
+//! Artifacts are written and fsynced first, re-read and digest-verified, and
+//! only then is the sealed manifest (completeness marker plus final digest)
+//! published. A crash at any earlier point leaves a bundle without a
+//! manifest, which decodes as an incomplete bundle and can never be restored.
+
+use crate::backends::local::{LocalKmsClient, StoredKeyProtection};
+use crate::backup::capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
+use crate::backup::error::BackupError;
+use crate::backup::manifest::{
+    AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
+    DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation,
+};
+use crate::error::{KmsError, Result};
+use aes_gcm::{
+    Aes256Gcm, Key, Nonce,
+    aead::{Aead, KeyInit, Payload},
+};
+use jiff::Zoned;
+use rand::RngExt;
+use serde::Deserialize;
+use std::path::{Path, PathBuf};
+use tokio::fs;
+use tokio::io::AsyncWriteExt;
+use zeroize::Zeroizing;
+
+/// File name of the sealed manifest inside a bundle directory.
+pub const LOCAL_BUNDLE_MANIFEST_FILE: &str = "manifest.json";
+const ARTIFACTS_DIR: &str = "artifacts";
+const KEYS_DIR: &str = "artifacts/keys";
+const SALT_ARTIFACT_PATH: &str = "artifacts/master-key.salt.enc";
+const AEAD_NONCE_LEN: usize = 12;
+/// Domain-separation context for the artifact AAD binding.
+const BUNDLE_AAD_CONTEXT: &str = "rustfs-kms-local-backup:v1";
+
+/// Caller-supplied backup KEK: a trust root deliberately separate from the
+/// business KMS hierarchy (it must not be a key that is itself part of the
+/// state being backed up). Where the KEK comes from is the admin layer's
+/// concern; this module only consumes it.
+pub struct BackupKek {
+    kek_id: String,
+    kek_version: u32,
+    key: Zeroizing<[u8; 32]>,
+}
+
+impl BackupKek {
+    /// Wrap 32 bytes of KEK material. The material is zeroized on drop;
+    /// callers should zeroize their own copy of the input.
+    pub fn new(kek_id: impl Into<String>, kek_version: u32, key: [u8; 32]) -> Result<Self> {
+        let kek_id = kek_id.into();
+        if kek_id.is_empty() {
+            return Err(KmsError::validation_error("backup KEK id must not be empty"));
+        }
+        Ok(Self {
+            kek_id,
+            kek_version,
+            key: Zeroizing::new(key),
+        })
+    }
+
+    /// Manifest descriptor for this KEK.
+    pub fn descriptor(&self) -> BackupKekDescriptor {
+        BackupKekDescriptor {
+            kek_id: self.kek_id.clone(),
+            kek_version: self.kek_version,
+            aead_algorithm: AeadAlgorithm::Aes256Gcm,
+        }
+    }
+
+    fn cipher(&self) -> Aes256Gcm {
+        Aes256Gcm::new(&Key::<Aes256Gcm>::from(*self.key))
+    }
+}
+
+/// Parameters of one export run.
+///
+/// `snapshot_generation` is injected by the caller: the contract only
+/// requires it to be monotonic per deployment, and the source (persisted
+/// counter, coordinated clock) is decided by the admin layer, which keeps
+/// this module free of ambient time or state lookups.
+#[derive(Debug, Clone)]
+pub struct LocalBackupExportRequest {
+    /// Unique identifier for this backup.
+    pub backup_id: String,
+    /// Opaque identity of the producing deployment.
+    pub deployment_identity: String,
+    /// RustFS version string recorded in the manifest.
+    pub rustfs_version: String,
+    /// Monotonic snapshot generation this bundle belongs to.
+    pub snapshot_generation: u64,
+    /// Bundle output directory; must not exist yet or must be empty.
+    pub destination: PathBuf,
+}
+
+impl LocalBackupExportRequest {
+    fn validate(&self) -> Result<()> {
+        for (field, value) in [
+            ("backup_id", &self.backup_id),
+            ("deployment_identity", &self.deployment_identity),
+            ("rustfs_version", &self.rustfs_version),
+        ] {
+            if value.is_empty() {
+                return Err(KmsError::validation_error(format!("backup export {field} must not be empty")));
+            }
+        }
+        Ok(())
+    }
+}
+
+/// Minimal projection of a stored key record: only the fields the exporter
+/// needs. Unknown fields are ignored on purpose — the record travels into the
+/// bundle byte-identical, so the exporter must not constrain its schema.
+#[derive(Deserialize)]
+struct StoredRecordProbe {
+    key_id: String,
+    #[serde(default)]
+    at_rest_protection: StoredKeyProtection,
+}
+
+struct CollectedRecord {
+    key_id: String,
+    protection: StoredKeyProtection,
+    /// Raw record bytes exactly as stored. Zeroized on drop because
+    /// plaintext-dev-only records embed key material.
+    raw: Zeroizing<Vec<u8>>,
+}
+
+struct CollectedSnapshot {
+    records: Vec<CollectedRecord>,
+    salt: Option<Vec<u8>>,
+}
+
+/// Export the Local backend's key directory as a sealed backup bundle.
+///
+/// The directory scan runs under the export fence, so concurrent
+/// create/update/delete operations are either fully included or fully
+/// excluded — never half a record. Encryption and bundle writing happen after
+/// the fence is released to keep it short.
+///
+/// Returns the sealed manifest that was written to the bundle.
+pub async fn export_local_backup(
+    client: &LocalKmsClient,
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+) -> Result<BackupManifest> {
+    request.validate()?;
+    prepare_destination(&request.destination).await?;
+
+    let snapshot = collect_snapshot(client).await?;
+    if snapshot.records.is_empty() {
+        return Err(KmsError::invalid_operation(
+            "Local backup export found no key records; refusing to publish an empty bundle",
+        ));
+    }
+
+    let has_encrypted = snapshot
+        .records
+        .iter()
+        .any(|record| record.protection == StoredKeyProtection::EncryptedMasterKey);
+    if has_encrypted && snapshot.salt.is_none() {
+        return Err(KmsError::invalid_operation(
+            "key directory contains encrypted-master-key records but the master key salt file is missing; \
+             the bundle would be unrestorable",
+        ));
+    }
+
+    let manifest = build_and_write_bundle(kek, request, &snapshot).await?;
+    Ok(manifest)
+}
+
+/// Read and fully validate the manifest of a local bundle directory.
+///
+/// A directory without a manifest is an interrupted export: the manifest is
+/// written last, so its absence means the bundle never sealed.
+pub async fn read_local_bundle_manifest(bundle_dir: &Path) -> Result<BackupManifest> {
+    let manifest_path = bundle_dir.join(LOCAL_BUNDLE_MANIFEST_FILE);
+    let bytes = match fs::read(&manifest_path).await {
+        Ok(bytes) => bytes,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Err(BackupError::incomplete_bundle("bundle has no manifest; the export never sealed it").into());
+        }
+        Err(error) => return Err(error.into()),
+    };
+    let manifest = BackupManifest::decode(&bytes)?;
+    if manifest.backend != BackupBackendKind::Local {
+        return Err(
+            BackupError::corrupted(format!("bundle manifest declares backend {:?}, expected Local", manifest.backend)).into(),
+        );
+    }
+    Ok(manifest)
+}
+
+/// Read, verify, and decrypt one artifact of a local bundle.
+///
+/// Fail-closed order: KEK identity, artifact presence, declared length,
+/// encrypted digest, then AEAD authentication. The returned plaintext is
+/// zeroized on drop.
+pub async fn decrypt_bundle_artifact(
+    bundle_dir: &Path,
+    manifest: &BackupManifest,
+    descriptor: &ArtifactDescriptor,
+    kek: &BackupKek,
+) -> Result<Zeroizing<Vec<u8>>> {
+    manifest.backup_kek.ensure_matches(&kek.kek_id, kek.kek_version)?;
+    if descriptor.aead_algorithm != AeadAlgorithm::Aes256Gcm {
+        return Err(KmsError::unsupported_algorithm(format!(
+            "{:?} (local bundles are produced with AES-256-GCM)",
+            descriptor.aead_algorithm
+        )));
+    }
+
+    let artifact_path = bundle_dir.join(&descriptor.path);
+    let payload = match fs::read(&artifact_path).await {
+        Ok(payload) => payload,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Err(BackupError::missing_artifact(descriptor.path.clone()).into());
+        }
+        Err(error) => return Err(error.into()),
+    };
+
+    if (payload.len() as u64) < descriptor.len {
+        return Err(BackupError::truncated(format!(
+            "artifact '{}' is {} bytes, manifest declares {}",
+            descriptor.path,
+            payload.len(),
+            descriptor.len
+        ))
+        .into());
+    }
+    if payload.len() as u64 != descriptor.len {
+        return Err(BackupError::corrupted(format!(
+            "artifact '{}' is {} bytes, manifest declares {}",
+            descriptor.path,
+            payload.len(),
+            descriptor.len
+        ))
+        .into());
+    }
+    if ContentDigest::sha256_of(&payload) != descriptor.encrypted_digest {
+        return Err(BackupError::corrupted(format!("artifact '{}' does not match its manifest digest", descriptor.path)).into());
+    }
+    if payload.len() < AEAD_NONCE_LEN {
+        return Err(BackupError::corrupted(format!("artifact '{}' is too short to carry a nonce", descriptor.path)).into());
+    }
+
+    let (nonce_bytes, ciphertext) = payload.split_at(AEAD_NONCE_LEN);
+    let mut nonce = [0u8; AEAD_NONCE_LEN];
+    nonce.copy_from_slice(nonce_bytes);
+    let aad = artifact_aad(&manifest.backup_id, manifest.snapshot_generation, &descriptor.path);
+    let plaintext = kek
+        .cipher()
+        .decrypt(
+            &Nonce::from(nonce),
+            Payload {
+                msg: ciphertext,
+                aad: &aad,
+            },
+        )
+        .map_err(|_| {
+            KmsError::from(BackupError::corrupted(format!(
+                "artifact '{}' failed authenticated decryption under the supplied backup KEK",
+                descriptor.path
+            )))
+        })?;
+    Ok(Zeroizing::new(plaintext))
+}
+
+/// Scan the key directory under the export fence.
+async fn collect_snapshot(client: &LocalKmsClient) -> Result<CollectedSnapshot> {
+    let _fence = client.acquire_export_fence().await;
+
+    let mut records = Vec::new();
+    let mut entries = fs::read_dir(client.key_directory()).await?;
+    while let Some(entry) = entries.next_entry().await? {
+        let path = entry.path();
+        if !path.extension().is_some_and(|extension| extension == "key") {
+            continue;
+        }
+        let stem = path
+            .file_stem()
+            .and_then(|stem| stem.to_str())
+            .ok_or_else(|| KmsError::configuration_error("Local KMS key file name must be valid UTF-8"))?
+            .to_string();
+
+        let raw = Zeroizing::new(fs::read(&path).await?);
+        // Any unreadable record aborts the export: a bundle silently missing
+        // one key is worse than no bundle at all.
+        let probe: StoredRecordProbe = serde_json::from_slice(&raw)
+            .map_err(|error| KmsError::material_corrupt(&stem, format!("stored key record does not deserialize: {error}")))?;
+        if probe.key_id != stem {
+            return Err(KmsError::invalid_key(format!(
+                "Local KMS key file identity mismatch: expected {stem:?}, found {:?}",
+                probe.key_id
+            )));
+        }
+
+        records.push(CollectedRecord {
+            key_id: stem,
+            protection: probe.at_rest_protection,
+            raw,
+        });
+    }
+
+    let salt_path = client.master_key_salt_file();
+    let salt = match fs::read(&salt_path).await {
+        Ok(bytes) => Some(bytes),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => None,
+        Err(error) => return Err(error.into()),
+    };
+
+    records.sort_by(|a, b| a.key_id.cmp(&b.key_id));
+    Ok(CollectedSnapshot { records, salt })
+}
+
+async fn build_and_write_bundle(
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+    snapshot: &CollectedSnapshot,
+) -> Result<BackupManifest> {
+    let mut artifacts = Vec::with_capacity(snapshot.records.len() + 1);
+    for record in &snapshot.records {
+        let artifact_path = format!("{KEYS_DIR}/{}.key.enc", record.key_id);
+        let descriptor = encrypt_and_write_artifact(kek, request, ArtifactKind::KeyMaterial, &artifact_path, &record.raw).await?;
+        artifacts.push(descriptor);
+    }
+    if let Some(salt) = &snapshot.salt {
+        let descriptor = encrypt_and_write_artifact(kek, request, ArtifactKind::MasterKeySalt, SALT_ARTIFACT_PATH, salt).await?;
+        artifacts.push(descriptor);
+    }
+
+    // Make the artifact directory entries durable before sealing: the sealed
+    // manifest must never survive a crash that its artifacts did not.
+    fsync_dir(&request.destination.join(KEYS_DIR)).await?;
+    fsync_dir(&request.destination.join(ARTIFACTS_DIR)).await?;
+
+    let manifest = BackupManifest {
+        format_version: BackupManifest::FORMAT_VERSION,
+        backup_id: request.backup_id.clone(),
+        // Normalized to UTC so the stored spelling is host-independent: the
+        // local zone's name (or a POSIX TZ string in minimal containers) has
+        // no business inside a portable bundle.
+        created_at: Zoned::now().with_time_zone(jiff::tz::TimeZone::UTC),
+        rustfs_version: request.rustfs_version.clone(),
+        deployment_identity: request.deployment_identity.clone(),
+        backend: BackupBackendKind::Local,
+        at_rest_protection: weakest_observed_protection(&snapshot.records),
+        responsibility: BackupResponsibility::FullMaterial,
+        snapshot_generation: request.snapshot_generation,
+        backup_kek: kek.descriptor(),
+        artifacts,
+        local_kdf: Some(local_kdf_descriptor(snapshot)),
+        key_versions: None,
+        capability_discovery: None,
+        completeness: CompletenessState::InProgress,
+        manifest_digest: ContentDigest {
+            algorithm: DigestAlgorithm::Sha256,
+            hex: String::new(),
+        },
+    };
+    let manifest = manifest.seal()?;
+    let manifest_bytes = manifest.encode()?;
+
+    write_new_file(&request.destination.join(LOCAL_BUNDLE_MANIFEST_FILE), &manifest_bytes).await?;
+    fsync_dir(&request.destination).await?;
+    Ok(manifest)
+}
+
+/// Encrypt one artifact, write it durably, and re-read it to verify the
+/// digest before it is allowed into the manifest.
+async fn encrypt_and_write_artifact(
+    kek: &BackupKek,
+    request: &LocalBackupExportRequest,
+    kind: ArtifactKind,
+    artifact_path: &str,
+    plaintext: &[u8],
+) -> Result<ArtifactDescriptor> {
+    let mut nonce = [0u8; AEAD_NONCE_LEN];
+    rand::rng().fill(&mut nonce[..]);
+    let aad = artifact_aad(&request.backup_id, request.snapshot_generation, artifact_path);
+    let ciphertext = kek
+        .cipher()
+        .encrypt(
+            &Nonce::from(nonce),
+            Payload {
+                msg: plaintext,
+                aad: &aad,
+            },
+        )
+        .map_err(|error| KmsError::cryptographic_error("backup_artifact_encrypt", error.to_string()))?;
+
+    let mut payload = Vec::with_capacity(AEAD_NONCE_LEN + ciphertext.len());
+    payload.extend_from_slice(&nonce);
+    payload.extend_from_slice(&ciphertext);
+
+    let absolute_path = request.destination.join(artifact_path);
+    write_new_file(&absolute_path, &payload).await?;
+
+    // Verify what actually landed on disk, not the in-memory buffer.
+    let written = fs::read(&absolute_path).await?;
+    let digest = ContentDigest::sha256_of(&written);
+    if written != payload {
+        return Err(KmsError::internal_error(format!(
+            "bundle artifact '{artifact_path}' read back differently than written"
+        )));
+    }
+
+    Ok(ArtifactDescriptor {
+        kind,
+        path: artifact_path.to_string(),
+        len: payload.len() as u64,
+        aead_algorithm: AeadAlgorithm::Aes256Gcm,
+        encrypted_digest: digest,
+    })
+}
+
+/// AAD binding an artifact to its bundle identity and path. A JSON tuple
+/// gives unambiguous field boundaries without a hand-rolled framing format.
+fn artifact_aad(backup_id: &str, snapshot_generation: u64, artifact_path: &str) -> Vec<u8> {
+    serde_json::to_vec(&(BUNDLE_AAD_CONTEXT, backup_id, snapshot_generation, artifact_path))
+        .expect("AAD tuple of strings and integers always serializes")
+}
+
+/// The bundle-level protection label is the weakest state observed across
+/// records: any plaintext-dev-only record marks the whole bundle, then any
+/// legacy-unspecified marker (unknown until read), and only a uniformly
+/// encrypted directory is labeled encrypted-master-key.
+fn weakest_observed_protection(records: &[CollectedRecord]) -> AtRestProtection {
+    let mut has_legacy = false;
+    for record in records {
+        match record.protection {
+            StoredKeyProtection::PlaintextDevOnly => return AtRestProtection::PlaintextDevOnly,
+            StoredKeyProtection::LegacyUnspecified => has_legacy = true,
+            StoredKeyProtection::EncryptedMasterKey => {}
+        }
+    }
+    if has_legacy {
+        AtRestProtection::LegacyUnspecified
+    } else {
+        AtRestProtection::EncryptedMasterKey
+    }
+}
+
+fn local_kdf_descriptor(snapshot: &CollectedSnapshot) -> LocalKdfDescriptor {
+    let mut modes = Vec::new();
+    for (marker, mode) in [
+        (StoredKeyProtection::EncryptedMasterKey, AtRestProtection::EncryptedMasterKey),
+        (StoredKeyProtection::PlaintextDevOnly, AtRestProtection::PlaintextDevOnly),
+        (StoredKeyProtection::LegacyUnspecified, AtRestProtection::LegacyUnspecified),
+    ] {
+        if snapshot.records.iter().any(|record| record.protection == marker) {
+            modes.push(mode);
+        }
+    }
+
+    // With a salt on disk the backend derives via Argon2id; without one only
+    // the pre-beta.9 SHA-256 derivation can apply. For plaintext-only
+    // directories the derivation is informational.
+    let derivation = if snapshot.salt.is_some() {
+        LocalKeyDerivation::current_argon2id()
+    } else {
+        LocalKeyDerivation::LegacySha256
+    };
+
+    LocalKdfDescriptor {
+        derivation,
+        protection_modes: modes,
+        // The verifier shape is left to the restore change; the schema keeps
+        // it optional so bundles without one stay valid.
+        master_key_verifier: None,
+    }
+}
+
+async fn prepare_destination(destination: &Path) -> Result<()> {
+    if fs::try_exists(destination).await? {
+        let mut entries = fs::read_dir(destination)
+            .await
+            .map_err(|error| KmsError::invalid_operation(format!("backup destination is not a readable directory: {error}")))?;
+        if entries.next_entry().await?.is_some() {
+            return Err(KmsError::invalid_operation(
+                "backup destination directory is not empty; refusing to mix bundles",
+            ));
+        }
+    }
+    fs::create_dir_all(destination.join(KEYS_DIR)).await?;
+    Ok(())
+}
+
+async fn write_new_file(path: &Path, bytes: &[u8]) -> Result<()> {
+    let mut file = fs::OpenOptions::new()
+        .write(true)
+        .create_new(true)
+        .open(path)
+        .await
+        .map_err(|error| KmsError::io_error(format!("failed to create bundle file {}: {error}", path.display())))?;
+    file.write_all(bytes).await?;
+    file.sync_all().await?;
+    Ok(())
+}
+
+/// Fsync a directory so freshly created bundle entries survive power loss.
+/// No-op on non-Unix platforms where directories cannot be opened for
+/// syncing (mirrors the local backend's durable commit helper).
+async fn fsync_dir(path: &Path) -> Result<()> {
+    #[cfg(unix)]
+    {
+        let path = path.to_path_buf();
+        tokio::task::spawn_blocking(move || std::fs::File::open(&path)?.sync_all())
+            .await
+            .map_err(|error| KmsError::io_error(error.to_string()))??;
+    }
+    #[cfg(not(unix))]
+    let _ = path;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::config::LocalConfig;
+    use std::sync::Arc;
+    use tempfile::TempDir;
+
+    async fn encrypted_client() -> (LocalKmsClient, TempDir) {
+        let temp = TempDir::new().expect("temp dir");
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp.path().to_path_buf(),
+            master_key: Some("test-master-key".to_string()),
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("client should initialize");
+        (client, temp)
+    }
+
+    async fn dev_client() -> (LocalKmsClient, TempDir) {
+        let temp = TempDir::new().expect("temp dir");
+        let client = LocalKmsClient::new(LocalConfig {
+            key_dir: temp.path().to_path_buf(),
+            master_key: None,
+            file_permissions: Some(0o600),
+        })
+        .await
+        .expect("client should initialize");
+        (client, temp)
+    }
+
+    fn test_kek() -> BackupKek {
+        BackupKek::new("backup-kek-test", 1, [0x42; 32]).expect("kek")
+    }
+
+    fn export_request(destination: PathBuf) -> LocalBackupExportRequest {
+        LocalBackupExportRequest {
+            backup_id: "backup-0001".to_string(),
+            deployment_identity: "deployment-test".to_string(),
+            rustfs_version: "1.0.0-test".to_string(),
+            snapshot_generation: 7,
+            destination,
+        }
+    }
+
+    fn walk_files(dir: &Path, out: &mut Vec<PathBuf>) {
+        for entry in std::fs::read_dir(dir).expect("read dir") {
+            let path = entry.expect("dir entry").path();
+            if path.is_dir() {
+                walk_files(&path, out);
+            } else {
+                out.push(path);
+            }
+        }
+    }
+
+    fn contains_subslice(haystack: &[u8], needle: &[u8]) -> bool {
+        !needle.is_empty() && haystack.windows(needle.len()).any(|window| window == needle)
+    }
+
+    #[tokio::test]
+    async fn export_round_trips_and_decrypts_to_source_records() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("alpha", "AES_256", None).await.expect("create alpha");
+        client.create_key("beta", "AES_256", None).await.expect("create beta");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.backend, BackupBackendKind::Local);
+        assert_eq!(manifest.responsibility, BackupResponsibility::FullMaterial);
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::EncryptedMasterKey);
+        assert_eq!(manifest.snapshot_generation, 7);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(kdf.derivation, LocalKeyDerivation::current_argon2id());
+        assert_eq!(kdf.protection_modes, vec![AtRestProtection::EncryptedMasterKey]);
+
+        // alpha, beta (sorted), then the salt artifact.
+        assert_eq!(manifest.artifacts.len(), 3);
+        assert_eq!(manifest.artifacts[0].path, "artifacts/keys/alpha.key.enc");
+        assert_eq!(manifest.artifacts[1].path, "artifacts/keys/beta.key.enc");
+        assert_eq!(manifest.artifacts[2].kind, ArtifactKind::MasterKeySalt);
+
+        let reread = read_local_bundle_manifest(&destination)
+            .await
+            .expect("manifest should decode");
+        assert_eq!(reread, manifest);
+
+        for (artifact, key_id) in [(&manifest.artifacts[0], "alpha"), (&manifest.artifacts[1], "beta")] {
+            let decrypted = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+                .await
+                .expect("artifact should decrypt");
+            let source = fs::read(client.key_directory().join(format!("{key_id}.key")))
+                .await
+                .expect("source record");
+            assert_eq!(decrypted.as_slice(), source.as_slice(), "record {key_id} must round-trip verbatim");
+        }
+
+        let salt = decrypt_bundle_artifact(&destination, &manifest, &manifest.artifacts[2], &kek)
+            .await
+            .expect("salt should decrypt");
+        let source_salt = fs::read(client.master_key_salt_file()).await.expect("source salt");
+        assert_eq!(salt.as_slice(), source_salt.as_slice());
+    }
+
+    #[tokio::test]
+    async fn plaintext_dev_only_material_is_rewrapped_and_absent_from_bundle() {
+        let (client, _key_dir) = dev_client().await;
+        client.create_key("dev-key", "AES_256", None).await.expect("create key");
+
+        let material = client
+            .decrypt_key_material_for_export("dev-key")
+            .await
+            .expect("material should be readable");
+        let source_record = fs::read(client.key_directory().join("dev-key.key")).await.expect("record");
+        let record_json: serde_json::Value = serde_json::from_slice(&source_record).expect("record parses");
+        let material_base64 = record_json
+            .get("encrypted_key_material")
+            .and_then(|value| value.as_str())
+            .expect("material field")
+            .to_string();
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::PlaintextDevOnly);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(kdf.protection_modes, vec![AtRestProtection::PlaintextDevOnly]);
+        assert_eq!(kdf.derivation, LocalKeyDerivation::LegacySha256);
+        assert!(
+            !manifest.artifacts.iter().any(|a| a.kind == ArtifactKind::MasterKeySalt),
+            "dev-mode directory has no salt to bundle"
+        );
+
+        // Byte-level: neither the raw material nor its base64 form may appear
+        // anywhere in the bundle. The mandatory KEK re-wrap is what hides it.
+        let mut files = Vec::new();
+        walk_files(&destination, &mut files);
+        assert!(!files.is_empty());
+        for file in files {
+            let bytes = std::fs::read(&file).expect("bundle file");
+            assert!(
+                !contains_subslice(&bytes, material.as_ref()),
+                "raw key material leaked into {}",
+                file.display()
+            );
+            assert!(
+                !contains_subslice(&bytes, material_base64.as_bytes()),
+                "base64 key material leaked into {}",
+                file.display()
+            );
+        }
+
+        // The wrapped record still round-trips for restore.
+        let decrypted = decrypt_bundle_artifact(&destination, &manifest, &manifest.artifacts[0], &kek)
+            .await
+            .expect("artifact should decrypt");
+        assert_eq!(decrypted.as_slice(), source_record.as_slice());
+    }
+
+    #[tokio::test]
+    async fn export_fence_blocks_writers_until_released() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("existing", "AES_256", None).await.expect("create key");
+        let client = Arc::new(client);
+
+        let fence = client.acquire_export_fence().await;
+
+        let writer = {
+            let client = Arc::clone(&client);
+            tokio::spawn(async move {
+                client.create_key("new-key", "AES_256", None).await.expect("create");
+                client.disable_key("existing", None).await.expect("disable");
+            })
+        };
+
+        for _ in 0..64 {
+            tokio::task::yield_now().await;
+        }
+        assert!(!writer.is_finished(), "writers must stay blocked while the export fence is held");
+
+        drop(fence);
+        writer.await.expect("writer should finish after fence release");
+        assert!(
+            fs::try_exists(client.key_directory().join("new-key.key"))
+                .await
+                .expect("exists")
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    async fn concurrent_writers_yield_complete_records() {
+        let (client, _key_dir) = encrypted_client().await;
+        for index in 0..5 {
+            client
+                .create_key(&format!("seed-{index}"), "AES_256", None)
+                .await
+                .expect("seed key");
+        }
+        let client = Arc::new(client);
+
+        let writer = {
+            let client = Arc::clone(&client);
+            tokio::spawn(async move {
+                for index in 0..30 {
+                    client
+                        .create_key(&format!("concurrent-{index}"), "AES_256", None)
+                        .await
+                        .expect("create");
+                    let target = format!("seed-{}", index % 5);
+                    if index % 2 == 0 {
+                        client.disable_key(&target, None).await.expect("disable");
+                    } else {
+                        client.enable_key(&target, None).await.expect("enable");
+                    }
+                }
+            })
+        };
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed under concurrent writers");
+        writer.await.expect("writer task");
+
+        // Whatever subset of writers landed before the fence, every record in
+        // the bundle must be complete: parseable, self-identifying, and with
+        // non-empty material. No torn records, no half-updates.
+        let reread = read_local_bundle_manifest(&destination).await.expect("manifest decodes");
+        assert_eq!(reread, manifest);
+        for artifact in manifest.artifacts.iter().filter(|a| a.kind == ArtifactKind::KeyMaterial) {
+            let record = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+                .await
+                .expect("record decrypts");
+            let value: serde_json::Value = serde_json::from_slice(&record).expect("record is complete JSON");
+            let key_id = value.get("key_id").and_then(|v| v.as_str()).expect("key_id present");
+            assert_eq!(artifact.path, format!("artifacts/keys/{key_id}.key.enc"));
+            let material = value
+                .get("encrypted_key_material")
+                .and_then(|v| v.as_str())
+                .expect("material present");
+            assert!(!material.is_empty());
+        }
+    }
+
+    #[tokio::test]
+    async fn tampered_and_truncated_bundles_fail_closed() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+        let artifact = &manifest.artifacts[0];
+        let artifact_file = destination.join(&artifact.path);
+        let original_artifact = std::fs::read(&artifact_file).expect("artifact bytes");
+
+        // Tampered artifact byte: digest verification rejects it.
+        let mut tampered = original_artifact.clone();
+        let last = tampered.len() - 1;
+        tampered[last] ^= 0x01;
+        std::fs::write(&artifact_file, &tampered).expect("write tampered");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+            .await
+            .expect_err("tampered artifact must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+
+        // Truncated artifact: typed truncation error.
+        std::fs::write(&artifact_file, &original_artifact[..original_artifact.len() - 4]).expect("truncate");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &kek)
+            .await
+            .expect_err("truncated artifact must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Truncated { .. })), "got {error:?}");
+        std::fs::write(&artifact_file, &original_artifact).expect("restore artifact");
+
+        // Tampered manifest (generation flip): sealed digest mismatch.
+        let manifest_file = destination.join(LOCAL_BUNDLE_MANIFEST_FILE);
+        let original_manifest = std::fs::read(&manifest_file).expect("manifest bytes");
+        let tampered_manifest = String::from_utf8(original_manifest.clone())
+            .expect("manifest is utf-8")
+            .replace("\"snapshot_generation\":7", "\"snapshot_generation\":8");
+        assert_ne!(tampered_manifest.as_bytes(), original_manifest.as_slice(), "tamper must apply");
+        std::fs::write(&manifest_file, tampered_manifest).expect("write tampered manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("tampered manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+
+        // Truncated manifest.
+        std::fs::write(&manifest_file, &original_manifest[..original_manifest.len() / 2]).expect("truncate manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("truncated manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Truncated { .. })), "got {error:?}");
+
+        // Missing manifest: the bundle never sealed.
+        std::fs::remove_file(&manifest_file).expect("remove manifest");
+        let error = read_local_bundle_manifest(&destination)
+            .await
+            .expect_err("bundle without manifest must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::IncompleteBundle { .. })), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn wrong_kek_is_rejected_before_decryption() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+        let artifact = &manifest.artifacts[0];
+
+        let wrong_id = BackupKek::new("other-kek", 1, [0x42; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_id)
+            .await
+            .expect_err("mismatched KEK id must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::WrongKek { .. })), "got {error:?}");
+
+        let wrong_version = BackupKek::new("backup-kek-test", 2, [0x42; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_version)
+            .await
+            .expect_err("mismatched KEK version must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::WrongKek { .. })), "got {error:?}");
+
+        // Right identity, wrong material: AEAD authentication fails closed.
+        let wrong_material = BackupKek::new("backup-kek-test", 1, [0x24; 32]).expect("kek");
+        let error = decrypt_bundle_artifact(&destination, &manifest, artifact, &wrong_material)
+            .await
+            .expect_err("wrong KEK material must be rejected");
+        assert!(matches!(error, KmsError::Backup(BackupError::Corrupted { .. })), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn missing_salt_with_encrypted_records_fails_export() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("victim", "AES_256", None).await.expect("create key");
+        fs::remove_file(client.master_key_salt_file()).await.expect("remove salt");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("export without salt must fail");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+        assert!(error.to_string().contains("salt"), "got {error}");
+    }
+
+    #[tokio::test]
+    async fn refuses_empty_key_dir_and_nonempty_destination() {
+        let (client, _key_dir) = dev_client().await;
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("empty key dir must not produce a bundle");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+
+        client.create_key("dev-key", "AES_256", None).await.expect("create key");
+        let occupied = bundle.path().join("occupied");
+        std::fs::create_dir_all(&occupied).expect("mkdir");
+        std::fs::write(occupied.join("stale"), b"leftover").expect("occupy");
+        let error = export_local_backup(&client, &test_kek(), &export_request(occupied))
+            .await
+            .expect_err("non-empty destination must be refused");
+        assert!(matches!(error, KmsError::InvalidOperation { .. }), "got {error:?}");
+    }
+
+    #[tokio::test]
+    async fn legacy_records_export_verbatim_with_weakest_protection_label() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("modern", "AES_256", None).await.expect("create key");
+        client.create_key("legacy-key", "AES_256", None).await.expect("create key");
+
+        // Strip the protection marker to fabricate a pre-beta.9 record, the
+        // same way the local backend's own legacy-compat tests do.
+        let legacy_path = client.key_directory().join("legacy-key.key");
+        let mut record: serde_json::Value =
+            serde_json::from_slice(&fs::read(&legacy_path).await.expect("record")).expect("record parses");
+        record
+            .as_object_mut()
+            .expect("record is an object")
+            .remove("at_rest_protection");
+        let legacy_bytes = serde_json::to_vec_pretty(&record).expect("record serializes");
+        fs::write(&legacy_path, &legacy_bytes).await.expect("write legacy record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let destination = bundle.path().join("bundle");
+        let kek = test_kek();
+        let manifest = export_local_backup(&client, &kek, &export_request(destination.clone()))
+            .await
+            .expect("export should succeed");
+
+        assert_eq!(manifest.at_rest_protection, AtRestProtection::LegacyUnspecified);
+        let kdf = manifest.local_kdf.as_ref().expect("local kdf descriptor");
+        assert_eq!(
+            kdf.protection_modes,
+            vec![AtRestProtection::EncryptedMasterKey, AtRestProtection::LegacyUnspecified]
+        );
+
+        let legacy_artifact = manifest
+            .artifacts
+            .iter()
+            .find(|a| a.path == "artifacts/keys/legacy-key.key.enc")
+            .expect("legacy artifact");
+        let decrypted = decrypt_bundle_artifact(&destination, &manifest, legacy_artifact, &kek)
+            .await
+            .expect("legacy artifact decrypts");
+        assert_eq!(decrypted.as_slice(), legacy_bytes.as_slice(), "legacy record must travel verbatim");
+    }
+
+    #[tokio::test]
+    async fn record_identity_mismatch_aborts_export() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("good", "AES_256", None).await.expect("create key");
+        std::fs::copy(client.key_directory().join("good.key"), client.key_directory().join("evil.key"))
+            .expect("plant mismatched record");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let error = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect_err("identity mismatch must abort the export");
+        assert!(matches!(error, KmsError::InvalidKey { .. }), "got {error:?}");
+    }
+}

crates/kms/src/backup/manifest.rs

@@ -355,8 +355,10 @@ struct ManifestProbe {
 ///
 /// The manifest is the authoritative description of one backup bundle: what
 /// was captured, under which snapshot generation, protected by which backup
-/// KEK, and which restore responsibility applies. Field order is part of the
-/// canonical digest form and is frozen for this format version.
+/// KEK, and which restore responsibility applies. The digest's canonical
+/// form is the manifest's JSON value with the digest hex emptied (see
+/// [`Self::compute_digest`]); decoders verify it against the raw stored
+/// bytes and never re-serialize parsed fields.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct BackupManifest {
@@ -415,8 +417,14 @@ impl BackupManifest {
     ///
     /// Fail-closed order: truncated or malformed input, then unknown format
     /// version, then a missing completeness marker, then schema decoding
-    /// (unknown fields, duplicate fields, missing fields), then semantic
-    /// validation including digest verification.
+    /// (unknown fields, duplicate fields, missing fields), then digest
+    /// verification against the raw input bytes, then semantic validation.
+    ///
+    /// The digest is verified against the bytes as stored — parsed typed
+    /// fields are never re-serialized for verification, so a field whose
+    /// string form does not round-trip byte-identically through its parsed
+    /// representation (timestamps in environment-dependent time zone
+    /// spellings, for example) cannot produce a spurious mismatch.
     pub fn decode(bytes: &[u8]) -> Result<Self, BackupError> {
         let probe: ManifestProbe = serde_json::from_slice(bytes).map_err(map_serde_error)?;
         if probe.format_version != Self::FORMAT_VERSION {
@@ -429,7 +437,9 @@ impl BackupManifest {
             return Err(BackupError::incomplete_bundle("manifest has no completeness marker"));
         }
         let manifest: Self = serde_json::from_slice(bytes).map_err(map_serde_error)?;
-        manifest.validate()?;
+        manifest.validate_pre_digest()?;
+        Self::verify_digest_in_bytes(bytes, &manifest.manifest_digest)?;
+        manifest.validate_content()?;
         Ok(manifest)
     }
 
@@ -449,23 +459,30 @@ impl BackupManifest {
         Ok(self)
     }
 
-    /// Compute the digest over the canonical manifest bytes.
+    /// Compute the digest over the canonical manifest form.
     ///
-    /// Canonical form: compact JSON serialization of this manifest with the
-    /// digest hex emptied. Field order is struct declaration order and is
-    /// frozen for format version 1, so the same manifest content always
-    /// hashes to the same value.
+    /// Canonical form: the JSON *value* of the manifest with the digest hex
+    /// emptied, object keys rebuilt in bytewise-sorted order at every level
+    /// (see [`canonicalize_value`]), then serialized compactly. The value
+    /// layer is what makes sealing and decoding agree byte-for-byte — a
+    /// decoder recovers the identical value from the raw stored bytes
+    /// without round-tripping any typed field through parse-and-reprint —
+    /// and the explicit key sort makes the bytes independent of
+    /// `serde_json`'s map implementation (`preserve_order` on or off).
     pub fn compute_digest(&self) -> Result<ContentDigest, BackupError> {
         let mut unsealed = self.clone();
         unsealed.manifest_digest = ContentDigest::placeholder(self.manifest_digest.algorithm);
-        let canonical = serde_json::to_vec(&unsealed)
+        let value = serde_json::to_value(&unsealed)
             .map_err(|error| BackupError::corrupted(format!("manifest canonicalization failed: {error}")))?;
-        match self.manifest_digest.algorithm {
-            DigestAlgorithm::Sha256 => Ok(ContentDigest::sha256_of(&canonical)),
-        }
+        Self::digest_of_canonical_value(value, self.manifest_digest.algorithm)
     }
 
-    /// Verify the sealed digest against the current manifest content.
+    /// Verify the sealed digest against the current in-memory content.
+    ///
+    /// This is the producer-side check (sealing and [`Self::encode`]).
+    /// Decoders must use the raw stored bytes instead (see [`Self::decode`]):
+    /// re-serializing parsed fields is not guaranteed to reproduce the
+    /// stored spelling byte-for-byte.
     pub fn verify_digest(&self) -> Result<(), BackupError> {
         if !self.manifest_digest.is_well_formed() {
             return Err(BackupError::corrupted("manifest digest is not a well-formed digest value"));
@@ -478,6 +495,33 @@ impl BackupManifest {
         Ok(())
     }
 
+    /// Verify a declared digest against raw manifest bytes, normalizing only
+    /// through the JSON value layer and emptying the digest slot in place.
+    fn verify_digest_in_bytes(bytes: &[u8], declared: &ContentDigest) -> Result<(), BackupError> {
+        if !declared.is_well_formed() {
+            return Err(BackupError::corrupted("manifest digest is not a well-formed digest value"));
+        }
+        let mut value: serde_json::Value = serde_json::from_slice(bytes).map_err(map_serde_error)?;
+        let Some(slot) = value.get_mut("manifest_digest").and_then(|digest| digest.get_mut("hex")) else {
+            return Err(BackupError::corrupted("manifest has no digest slot"));
+        };
+        *slot = serde_json::Value::String(String::new());
+        if Self::digest_of_canonical_value(value, declared.algorithm)? != *declared {
+            return Err(BackupError::corrupted(
+                "manifest digest mismatch: content does not match the sealed digest",
+            ));
+        }
+        Ok(())
+    }
+
+    fn digest_of_canonical_value(value: serde_json::Value, algorithm: DigestAlgorithm) -> Result<ContentDigest, BackupError> {
+        let canonical = serde_json::to_vec(&canonicalize_value(value))
+            .map_err(|error| BackupError::corrupted(format!("manifest canonicalization failed: {error}")))?;
+        match algorithm {
+            DigestAlgorithm::Sha256 => Ok(ContentDigest::sha256_of(&canonical)),
+        }
+    }
+
     /// Look up a required artifact by kind, failing closed when absent.
     pub fn require_artifact(&self, kind: ArtifactKind) -> Result<&ArtifactDescriptor, BackupError> {
         self.artifacts
@@ -486,11 +530,21 @@ impl BackupManifest {
             .ok_or_else(|| BackupError::missing_artifact(artifact_kind_name(kind)))
     }
 
-    /// Validate the full manifest contract.
+    /// Validate the full manifest contract against the in-memory content.
     ///
-    /// This is decode-side validation and also guards [`Self::encode`], so a
-    /// producer cannot publish a manifest a decoder would reject.
+    /// This guards [`Self::encode`], so a producer cannot publish a manifest
+    /// a decoder would reject. [`Self::decode`] runs the same checks but
+    /// verifies the digest against the raw input bytes instead.
     pub fn validate(&self) -> Result<(), BackupError> {
+        self.validate_pre_digest()?;
+        self.verify_digest()?;
+        self.validate_content()
+    }
+
+    /// Checks that must run before any digest verification: an unknown
+    /// version or an unsealed bundle is reported as its own typed error, not
+    /// as a digest mismatch.
+    fn validate_pre_digest(&self) -> Result<(), BackupError> {
         if self.format_version != Self::FORMAT_VERSION {
             return Err(BackupError::UnknownVersion {
                 found: self.format_version,
@@ -500,7 +554,12 @@ impl BackupManifest {
         if self.completeness != CompletenessState::Complete {
             return Err(BackupError::incomplete_bundle("completeness marker records an in-progress bundle"));
         }
-        self.verify_digest()?;
+        Ok(())
+    }
+
+    /// Semantic validation of everything except version, completeness, and
+    /// digest integrity.
+    fn validate_content(&self) -> Result<(), BackupError> {
         require_non_empty("backup_id", &self.backup_id)?;
         require_non_empty("rustfs_version", &self.rustfs_version)?;
         require_non_empty("deployment_identity", &self.deployment_identity)?;
@@ -645,6 +704,29 @@ fn map_serde_error(error: serde_json::Error) -> BackupError {
     }
 }
 
+/// Rebuild a JSON value with object keys in bytewise-sorted order at every
+/// nesting level (array element order is preserved).
+///
+/// `serde_json`'s map keeps keys sorted by default but preserves insertion
+/// order when the `preserve_order` feature is unified into the build by any
+/// other crate. Digest bytes must not depend on that, so the ordering is
+/// imposed explicitly here instead of being inherited from the map type.
+fn canonicalize_value(value: serde_json::Value) -> serde_json::Value {
+    match value {
+        serde_json::Value::Object(map) => {
+            let mut entries: Vec<(String, serde_json::Value)> = map.into_iter().collect();
+            entries.sort_by(|a, b| a.0.cmp(&b.0));
+            let mut sorted = serde_json::Map::with_capacity(entries.len());
+            for (key, entry) in entries {
+                sorted.insert(key, canonicalize_value(entry));
+            }
+            serde_json::Value::Object(sorted)
+        }
+        serde_json::Value::Array(items) => serde_json::Value::Array(items.into_iter().map(canonicalize_value).collect()),
+        other => other,
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -736,7 +818,7 @@ mod tests {
     /// canonical form and frozen. If serialization layout or field order
     /// changes, this value changes and the fixture test fails — which is the
     /// point: that is a format-version bump, not a patch.
-    const FIXTURE_DIGEST_HEX: &str = "01accb3e2bc51e12d17d1efc52ad6ab9441c50bec52c3fb29e0a9f92b725cdaa";
+    const FIXTURE_DIGEST_HEX: &str = "a8b104d61c358cd75cc9a7691d2f7d2d96f73c53653bef8940a49e96dbdf7775";
 
     fn fixture() -> String {
         FIXTURE.replace("SEALED_DIGEST_HEX", FIXTURE_DIGEST_HEX)
@@ -1010,12 +1092,39 @@ mod tests {
         let with_discovery = fixture().replace("\"completeness\"", "\"capability_discovery\": [1], \"completeness\"");
         expect_corrupted(BackupManifest::decode(with_discovery.as_bytes()), "reserved");
 
-        // Explicit null carries no data and is tolerated as absence; digest
-        // verification still passes because null slots are skipped on
-        // serialization.
+        // An explicit null slot decodes as absence at the schema layer, but
+        // sealed bundles never contain the key (`skip_serializing_if`), so
+        // inserting one after sealing is a byte-level modification and the
+        // raw-bytes digest check rejects it.
         let with_null = fixture().replace("\"completeness\"", "\"key_versions\": null, \"completeness\"");
-        let decoded = BackupManifest::decode(with_null.as_bytes()).expect("null reserved slot should decode");
-        assert_eq!(decoded.key_versions, None);
+        expect_corrupted(BackupManifest::decode(with_null.as_bytes()), "digest mismatch");
+    }
+
+    #[test]
+    fn digest_verification_survives_non_round_tripping_timestamp_spellings() {
+        // A legacy `created_at` spelling (no time zone annotation) parses via
+        // the compat fallback and re-serializes differently ("+00:00[UTC]"),
+        // and host-dependent zone spellings can do the same. Digest
+        // verification therefore operates on the raw stored bytes and must
+        // never re-serialize parsed fields.
+        let sealed = seal(local_manifest_unsealed());
+        let mut value = serde_json::to_value(&sealed).expect("manifest should convert to a value");
+        value["created_at"] = serde_json::Value::String("2026-07-30T00:00:00+00:00".to_string());
+        value["manifest_digest"]["hex"] = serde_json::Value::String(String::new());
+        let digest = BackupManifest::digest_of_canonical_value(value.clone(), DigestAlgorithm::Sha256)
+            .expect("canonical digest should compute");
+        value["manifest_digest"]["hex"] = serde_json::Value::String(digest.hex);
+        let bytes = serde_json::to_vec(&value).expect("manifest bytes");
+
+        let decoded = BackupManifest::decode(&bytes).expect("a non-round-tripping timestamp spelling must not break decoding");
+
+        // Precondition: the spelling really does not survive a typed
+        // round-trip — otherwise this test is vacuous.
+        let reserialized = serde_json::to_value(&decoded).expect("decoded manifest should convert to a value");
+        assert_ne!(reserialized["created_at"], value["created_at"]);
+        // Which is exactly why the producer-side (in-memory) digest check
+        // cannot be used on decoded manifests.
+        assert!(decoded.verify_digest().is_err());
     }
 
     #[test]

crates/kms/src/backup/mod.rs

@@ -12,13 +12,13 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Backup/restore contract types for KMS state.
+//! Backup/restore contracts and backup production for KMS state.
 //!
-//! This module is contract-only: it defines the versioned backup manifest,
-//! the per-backend responsibility matrix, typed failure modes, and the
-//! restore dry-run report. Nothing here is wired into handlers or backends;
-//! backup export, restore orchestration, and the admin API build on these
-//! types in follow-up changes.
+//! The contract side defines the versioned backup manifest, the per-backend
+//! responsibility matrix, typed failure modes, and the restore dry-run
+//! report. [`local_export`] implements the producer side for the Local
+//! backend as a crate-internal API; restore orchestration and the admin API
+//! build on these pieces in follow-up changes.
 //!
 //! # Bundle model
 //!
@@ -50,6 +50,7 @@
 mod capability;
 mod dry_run;
 mod error;
+pub mod local_export;
 mod manifest;
 
 pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
@@ -57,6 +58,10 @@ pub use dry_run::{
     ExternalDependencyMismatch, RestoreBlocker, RestoreBlockerCode, RestoreConflict, RestoreConflictKind, RestoreDryRunReport,
 };
 pub use error::BackupError;
+pub use local_export::{
+    BackupKek, LOCAL_BUNDLE_MANIFEST_FILE, LocalBackupExportRequest, decrypt_bundle_artifact, export_local_backup,
+    read_local_bundle_manifest,
+};
 pub use manifest::{
     AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
     DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation, ReservedSlot,

crates/kms/tests/vault_fault_injection.rs

@@ -33,9 +33,11 @@ use std::time::Duration;
 
 use metrics_util::MetricKind;
 use metrics_util::debugging::{DebugValue, DebuggingRecorder};
-use rustfs_kms::backends::KmsClient;
-use rustfs_kms::backends::vault::VaultKmsClient;
-use rustfs_kms::{KmsConfig, KmsError, VaultAuthMethod, VaultConfig};
+use rustfs_kms::backends::KmsBackend as KmsBackendTrait;
+use rustfs_kms::backends::vault::VaultKmsBackend;
+use rustfs_kms::{
+    BackendConfig, DescribeKeyRequest, KmsBackend as KmsBackendKind, KmsConfig, KmsError, VaultAuthMethod, VaultConfig,
+};
 
 const OPERATIONS_TOTAL: &str = "rustfs_kms_backend_operations_total";
 const ATTEMPT_FAILURES_TOTAL: &str = "rustfs_kms_backend_attempt_failures_total";
@@ -54,14 +56,23 @@ fn vault_config(address: &str, token: &str) -> VaultConfig {
     }
 }
 
-fn kms_config(attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
+fn kms_config(vault_config: VaultConfig, attempt_timeout: Duration, retry_attempts: u32) -> KmsConfig {
     KmsConfig {
+        backend: KmsBackendKind::VaultKv2,
+        backend_config: BackendConfig::VaultKv2(Box::new(vault_config)),
+        allow_insecure_dev_defaults: true,
         timeout: attempt_timeout,
         retry_attempts,
         ..KmsConfig::default()
     }
 }
 
+fn describe_key_request(key_id: &str) -> DescribeKeyRequest {
+    DescribeKeyRequest {
+        key_id: key_id.to_string(),
+    }
+}
+
 type MetricEntry = (
     metrics_util::CompositeKey,
     Option<metrics::Unit>,
@@ -118,11 +129,10 @@ fn connection_refused_is_retried_within_budget() {
 
     let snapshot = record_metrics(|| {
         Box::pin(async move {
-            let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_secs(2), 2))
+            let client = VaultKmsBackend::new(kms_config(vault_config(&address, "unused"), Duration::from_secs(2), 2))
                 .await
                 .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-refused", None)
+            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-refused"))
                 .await
                 .expect_err("a refused connection must fail the operation");
             assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
@@ -166,11 +176,10 @@ fn stalled_connection_is_cut_off_by_the_attempt_timeout() {
                 }
             });
 
-            let client = VaultKmsClient::new(vault_config(&address, "unused"), &kms_config(Duration::from_millis(250), 1))
+            let client = VaultKmsBackend::new(kms_config(vault_config(&address, "unused"), Duration::from_millis(250), 1))
                 .await
                 .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-stalled", None)
+            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-stalled"))
                 .await
                 .expect_err("a stalled request must be cut off by the attempt timeout");
             assert!(
@@ -201,11 +210,10 @@ fn real_vault_invalid_token_is_fatal_and_never_retried() {
     let snapshot = record_metrics(|| {
         Box::pin(async {
             let config = vault_config(&real_vault_address(), "fault-injection-invalid-token");
-            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
+            let client = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3))
                 .await
                 .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-forbidden", None)
+            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-forbidden"))
                 .await
                 .expect_err("an invalid token must be rejected");
             assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
@@ -235,11 +243,10 @@ fn real_vault_missing_key_is_resolved_in_one_attempt() {
     let snapshot = record_metrics(|| {
         Box::pin(async move {
             let config = vault_config(&real_vault_address(), &token);
-            let client = VaultKmsClient::new(config, &kms_config(Duration::from_secs(5), 3))
+            let client = VaultKmsBackend::new(kms_config(config, Duration::from_secs(5), 3))
                 .await
                 .expect("client construction performs no network calls");
-            let error = client
-                .describe_key("fault-injection-definitely-missing", None)
+            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-definitely-missing"))
                 .await
                 .expect_err("a missing key must resolve to key-not-found");
             assert!(matches!(error, KmsError::KeyNotFound { .. }), "got {error:?}");

docs/operations/hotpath-warp-abba-runbook.md

@@ -0,0 +1,277 @@
+# Hotpath warp ABBA validation runbook
+
+This runbook describes how to collect formal Linux or production-cluster
+evidence for hotpath performance changes. Use it when a short local A/B smoke
+run is too noisy to decide whether a regression is real.
+
+The ABBA runner executes each workload and drive-sync cell as:
+
+```text
+A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
+```
+
+`B1` and `B2` are compared with `A1` to measure the candidate delta. `A2` is
+also compared with `A1` to measure baseline drift. Treat a candidate regression
+as actionable only when the `A2` drift is passing or materially smaller than
+the `B1` and `B2` delta for the same workload.
+
+## Scope
+
+Use this runbook for hotpath profiling and performance validation of RustFS
+object I/O changes, especially when CPU, memory allocation, lock/channel wait
+time, request throughput, or tail latency is the review question.
+
+The script validates the same workload matrix as the hotpath warp A/B gate:
+
+| Workload | mode | size |
+| --- | --- | --- |
+| `put-4kib` | put | 4KiB |
+| `put-4mib` | put | 4MiB |
+| `get-4kib` | get | 4KiB |
+| `get-4mib` | get | 4MiB |
+| `get-10mib` | get | 10MiB |
+| `mixed-256k` | mixed | 256KiB |
+
+Each workload runs with `RUSTFS_DRIVE_SYNC_ENABLE=true` and
+`RUSTFS_DRIVE_SYNC_ENABLE=false`, so a full ABBA pass produces 48 measurement
+cells: 6 workloads x 2 drive-sync modes x 4 ABBA legs.
+
+## Prerequisites
+
+Run the formal pass on Linux, not on a laptop smoke environment.
+
+Required tools on the bench host:
+
+- `bash`, `curl`, `git`, and core GNU userland.
+- `warp` on `PATH`, or pass `--warp-bin`.
+- Two RustFS Linux binaries: one baseline and one candidate.
+- Enough isolated disks or directories for the local runner, or an externally
+  managed RustFS cluster for production-like validation.
+- Stable host telemetry collection such as `pidstat`, `mpstat`, `iostat`,
+  `sar`, `perf`, `heaptrack`, or the platform's equivalent observability stack.
+
+Cluster-mode requirements:
+
+- A deploy hook that can replace the RustFS binary on every node.
+- The hook must apply `RUSTFS_DRIVE_SYNC_ENABLE` for the current ABBA leg.
+- The hook must restart RustFS and return only after the rollout command has
+  been accepted. The ABBA script performs the HTTP readiness wait.
+- The benchmark client should run outside the RustFS nodes when possible.
+- Do not run against a production data set unless the workload bucket and test
+  credentials are isolated and approved for destructive benchmark traffic.
+
+## Build the binaries
+
+Build the baseline from the comparison commit, usually `origin/main` or the
+previous accepted release:
+
+```bash
+git fetch origin main
+git switch --detach origin/main
+cargo build --release -p rustfs --bins
+cp target/release/rustfs /tmp/rustfs-baseline
+```
+
+Build the candidate from the PR commit:
+
+```bash
+git switch <candidate-branch>
+cargo build --release -p rustfs --bins
+cp target/release/rustfs /tmp/rustfs-candidate
+```
+
+For cross-compiled cluster binaries, keep both outputs on the bench host and
+make the deploy hook copy the selected binary to the cluster. The ABBA runner
+passes the selected binary path through `HOTPATH_ABBA_BINARY`.
+
+## Local Linux runner
+
+Use local mode for a dedicated Linux runner with disposable data paths. This is
+not a substitute for a production-like cluster, but it is useful before spending
+cluster time.
+
+```bash
+scripts/run_hotpath_warp_abba.sh \
+  --baseline-bin /tmp/rustfs-baseline \
+  --candidate-bin /tmp/rustfs-candidate \
+  --address 127.0.0.1:9000 \
+  --data-root /var/tmp/rustfs-hotpath-abba \
+  --disks 4 \
+  --duration 120s \
+  --rounds 3 \
+  --cooldown 30 \
+  --concurrency 16 \
+  --out-dir target/hotpath-abba/linux-local
+```
+
+The script starts and stops RustFS for each ABBA leg. The data root is
+throwaway and should not contain important data.
+
+## Production-like cluster runner
+
+Use external mode when RustFS lifecycle is managed by ansible, systemd, a
+cluster scheduler, or a dedicated deployment harness. In this mode the ABBA
+script does not start RustFS directly; it calls `--deploy-hook` before each leg
+and then waits for `http://<endpoint><health-path>`.
+
+The deploy hook receives:
+
+| Environment variable | Value |
+| --- | --- |
+| `HOTPATH_ABBA_LEG` | `A1`, `B1`, `B2`, or `A2` |
+| `HOTPATH_ABBA_PHASE` | `baseline` or `candidate` |
+| `HOTPATH_ABBA_BINARY` | selected baseline or candidate binary path |
+| `HOTPATH_ABBA_DRIVE_SYNC` | `true` or `false` |
+
+Example ansible-shaped command:
+
+```bash
+scripts/run_hotpath_warp_abba.sh \
+  --baseline-bin /srv/rustfs-binaries/rustfs-baseline \
+  --candidate-bin /srv/rustfs-binaries/rustfs-candidate \
+  --endpoint rustfs-bench.example.internal:9000 \
+  --deploy-hook '
+    set -euo pipefail
+    cd /srv/rustfs-ansible
+    cp "${HOTPATH_ABBA_BINARY:?}" roles/rustfs/files/rustfs
+    export RUSTFS_DRIVE_SYNC_ENABLE="${HOTPATH_ABBA_DRIVE_SYNC:?}"
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags stop
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags config
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags binary-copy
+    ansible-playbook -f 4 -l bench rustfs-manage.yml --tags start
+  ' \
+  --duration 180s \
+  --rounds 5 \
+  --cooldown 45 \
+  --concurrency 32 \
+  --out-dir target/hotpath-abba/cluster-pr-XXXX
+```
+
+For formal evidence, prefer `--rounds 5` or higher when the cluster budget
+allows it. The script enforces `--rounds >= 3`.
+
+## CPU and memory evidence
+
+ABBA warp output answers whether the candidate changed throughput or latency.
+Collect host telemetry at the same time to explain why.
+
+Recommended minimum:
+
+```bash
+mkdir -p target/hotpath-abba/cluster-pr-XXXX/telemetry
+
+pidstat -durh 5 > target/hotpath-abba/cluster-pr-XXXX/telemetry/pidstat.txt &
+PIDSTAT_PID=$!
+
+mpstat 5 > target/hotpath-abba/cluster-pr-XXXX/telemetry/mpstat.txt &
+MPSTAT_PID=$!
+
+iostat -xz 5 > target/hotpath-abba/cluster-pr-XXXX/telemetry/iostat.txt &
+IOSTAT_PID=$!
+```
+
+Stop the collectors after the ABBA script exits:
+
+```bash
+kill "$PIDSTAT_PID" "$MPSTAT_PID" "$IOSTAT_PID"
+```
+
+For deeper CPU attribution, run `perf record` around one representative
+workload after the ABBA gate identifies a candidate regression or improvement:
+
+```bash
+perf record -F 99 -g -- sleep 180
+perf report --stdio > target/hotpath-abba/cluster-pr-XXXX/telemetry/perf-report.txt
+```
+
+For allocation profiling, build the candidate with:
+
+```bash
+cargo build --release -p rustfs --bins --features hotpath-alloc
+```
+
+Then run the same ABBA command with that binary. Compare allocation-heavy
+function sections only within the same build mode. Do not compare
+`hotpath-alloc` binaries directly with default release binaries for throughput
+acceptance, because allocation instrumentation intentionally changes what is
+measured.
+
+For CPU hotpath sections emitted by hotpath, build with:
+
+```bash
+cargo build --release -p rustfs --bins --features hotpath-cpu
+```
+
+Use the CPU-enabled report to explain hotspots after the default or plain
+`hotpath` ABBA gate shows a real effect.
+
+## Output layout
+
+The ABBA runner writes:
+
+```text
+<out-dir>/
+  manifest.env
+  abba_schedule.csv
+  candidate_gate.md
+  baseline_drift_gate.md
+  summary.md
+  <workload>/<sync>/<leg>/median_summary.csv
+  <workload>/<sync>/<leg>/baseline_compare.csv
+```
+
+Attach or link at least these files in the issue or PR:
+
+- `summary.md`
+- `candidate_gate.md`
+- `baseline_drift_gate.md`
+- `abba_schedule.csv`
+- every `median_summary.csv` and `baseline_compare.csv` for a failed or
+  borderline workload
+- host telemetry files used to explain CPU, memory, or disk saturation
+
+## Interpretation
+
+Use this decision table:
+
+| Candidate gate | A2 drift gate | Interpretation |
+| --- | --- | --- |
+| PASS | PASS | Candidate is acceptable for the measured matrix. |
+| WARN | PASS | Candidate has a small measurable signal; inspect telemetry and decide if it is expected. |
+| FAIL | PASS | Candidate likely regressed the affected workload; investigate before merge. |
+| FAIL | FAIL on the same workload | Environment drift is high; rerun on a quieter runner or increase duration and rounds. |
+| PASS | FAIL | Candidate did not exceed the budget, but the rig was unstable; avoid using the numbers as proof of improvement. |
+
+When `B1` and `B2` disagree, treat the result as inconclusive even if the gate
+passes. Increase duration, rounds, cooldown, or runner isolation before drawing
+a conclusion.
+
+## AI execution checklist
+
+When delegating the run to an AI agent or an automation runner, provide these
+inputs explicitly:
+
+- repository checkout and candidate branch or commit;
+- baseline commit or binary path;
+- candidate binary path;
+- runner type: local Linux or external cluster;
+- endpoint, access key, secret key source, and region;
+- deploy hook path or exact command for cluster mode;
+- output directory;
+- required duration, rounds, cooldown, concurrency, and fail/warn budgets;
+- where to upload artifacts after the run.
+
+The AI agent should execute this sequence:
+
+1. Confirm `uname -a`, RustFS commits, binary SHA256 sums, `warp --version`,
+   CPU model, memory size, disk layout, and whether the run is local or cluster.
+2. Run `scripts/run_hotpath_warp_abba.sh --dry-run` with the final arguments.
+3. Run the real ABBA command with `--rounds >= 3`.
+4. Preserve the full output directory without editing generated CSV files.
+5. Read `summary.md`, `candidate_gate.md`, and `baseline_drift_gate.md`.
+6. Summarize only measured facts: candidate deltas, baseline drift, CPU or
+   memory saturation, and any failed workloads.
+7. Post the summary and artifact location to the tracking issue or PR.
+
+Do not report a performance win or loss when the baseline drift gate failed on
+the same workload and no rerun was collected.

rustfs/src/admin/router.rs

@@ -2708,18 +2708,16 @@ impl<T: Operation> S3Router<T> {
 
     pub fn insert(&mut self, method: Method, path: &str, operation: T) -> std::io::Result<()> {
         let path = Self::make_route_str(method, path);
+        #[cfg(test)]
+        let registered_path = path.clone();
 
         // warn!("set uri {}", &path);
 
-        #[cfg(test)]
-        {
-            self.router.insert(path.clone(), operation).map_err(std::io::Error::other)?;
-            self.registered_routes.push(path);
-        }
-
-        #[cfg(not(test))]
         self.router.insert(path, operation).map_err(std::io::Error::other)?;
 
+        #[cfg(test)]
+        self.registered_routes.push(registered_path);
+
         Ok(())
     }
 

scripts/run_hotpath_warp_abba.sh

@@ -0,0 +1,428 @@
+#!/usr/bin/env bash
+# Formal Linux / production-cluster ABBA runner for the hotpath warp matrix.
+#
+# This script is intentionally a thin orchestrator around the existing
+# run_object_batch_bench_enhanced.sh load driver and hotpath_warp_ab_gate.sh
+# relative-budget gate. It runs each durability/workload cell as:
+#
+#   A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
+#
+# Candidate legs are compared against A1. The final A2 leg is also compared
+# against A1 to quantify baseline drift separately from candidate deltas.
+
+set -euo pipefail
+
+PROJECT_ROOT="$(git rev-parse --show-toplevel)"
+ENHANCED_BENCH="${PROJECT_ROOT}/scripts/run_object_batch_bench_enhanced.sh"
+GATE="${PROJECT_ROOT}/scripts/hotpath_warp_ab_gate.sh"
+
+BASELINE_BIN=""
+CANDIDATE_BIN=""
+ENDPOINT=""
+DEPLOY_HOOK=""
+HEALTH_PATH="/health"
+ADDRESS="127.0.0.1:9000"
+DATA_ROOT="/tmp/rustfs-hotpath-abba"
+DISKS=4
+ACCESS_KEY="rustfsadmin"
+SECRET_KEY="rustfsadmin"
+REGION="us-east-1"
+WARP_BIN="warp"
+CONCURRENCY=8
+DURATION="60s"
+ROUNDS=3
+COOLDOWN_SECS=20
+HEALTH_TIMEOUT_SECS=180
+FAIL_PCT=10
+WARN_PCT=5
+ALLOW_REGRESSION=false
+EXEMPTION_REASON="deliberate correctness tradeoff"
+OUT_DIR="${PROJECT_ROOT}/target/hotpath-abba/$(date -u +%Y%m%dT%H%M%SZ 2>/dev/null || echo run)"
+DRY_RUN=false
+
+WORKLOADS=(
+  "put-4kib|put|4KiB"
+  "put-4mib|put|4MiB"
+  "get-4kib|get|4KiB"
+  "get-4mib|get|4MiB"
+  "get-10mib|get|10MiB"
+  "mixed-256k|mixed|256KiB"
+)
+DRIVE_SYNC_MATRIX=("sync-on|true" "sync-off|false")
+
+usage() {
+  cat <<'USAGE'
+Usage: scripts/run_hotpath_warp_abba.sh --baseline-bin <path> --candidate-bin <path> [options]
+
+Formal ABBA mode for Linux runners or production-like clusters. The schedule is
+A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline for every workload
+and drive-sync cell.
+
+Required:
+  --baseline-bin <path>     Baseline RustFS binary.
+  --candidate-bin <path>    Candidate RustFS binary.
+
+Local Linux runner mode:
+  --address <host:port>     Local RustFS address (default 127.0.0.1:9000).
+  --disks <n>               Throwaway local disks per node (default 4).
+  --data-root <path>        Local disk root (default /tmp/rustfs-hotpath-abba).
+
+Production / cluster mode:
+  --endpoint <host:port>    Existing cluster endpoint. Enables external mode.
+  --deploy-hook <cmd>       Command run before each ABBA leg. It receives:
+                              HOTPATH_ABBA_LEG=A1|B1|B2|A2
+                              HOTPATH_ABBA_PHASE=baseline|candidate
+                              HOTPATH_ABBA_BINARY=<baseline/candidate binary>
+                              HOTPATH_ABBA_DRIVE_SYNC=true|false
+  --health-path <path>      Readiness path (default /health).
+
+Benchmark:
+  --duration <dur>          warp duration per cell (default 60s).
+  --rounds <n>              rounds per cell; must be >= 3 (default 3).
+  --cooldown <n>            cooldown seconds between rounds/sizes (default 20).
+  --concurrency <n>         warp concurrency (default 8).
+  --warp-bin <path>         warp binary (default warp).
+
+Credentials:
+  --access-key <value>      S3 access key (default rustfsadmin).
+  --secret-key <value>      S3 secret key (default rustfsadmin).
+  --region <value>          S3 region (default us-east-1).
+
+Gate:
+  --fail-pct <n>            Regression budget that fails gate (default 10).
+  --warn-pct <n>            Regression budget that warns (default 5).
+  --allow-regression        Downgrade candidate gate FAIL to WARN.
+  --exemption-reason <s>    Reason recorded when allow-regression is used.
+
+Output:
+  --out-dir <path>          Output dir (default target/hotpath-abba/<ts>).
+  --dry-run                 Print commands without starting servers or warp.
+  -h, --help
+
+Outputs:
+  <out-dir>/abba_schedule.csv
+  <out-dir>/candidate_gate.md
+  <out-dir>/baseline_drift_gate.md
+  <out-dir>/summary.md
+  <out-dir>/<workload>/<sync>/<leg>/{median_summary.csv,baseline_compare.csv}
+USAGE
+}
+
+die() {
+  echo "error: $*" >&2
+  exit 2
+}
+
+log() {
+  printf '[hotpath-abba] %s\n' "$*" >&2
+}
+
+run() {
+  if [[ "$DRY_RUN" == "true" ]]; then
+    { printf 'DRY-RUN:'; printf ' %q' "$@"; printf '\n'; } >&2
+    return 0
+  fi
+  "$@"
+}
+
+validate_positive_int() {
+  local value="$1" name="$2"
+  [[ "$value" =~ ^[0-9]+$ && "$value" -gt 0 ]] || die "$name must be a positive integer"
+}
+
+while [[ $# -gt 0 ]]; do
+  case "$1" in
+    --baseline-bin) BASELINE_BIN="$2"; shift 2 ;;
+    --candidate-bin) CANDIDATE_BIN="$2"; shift 2 ;;
+    --endpoint) ENDPOINT="$2"; shift 2 ;;
+    --deploy-hook) DEPLOY_HOOK="$2"; shift 2 ;;
+    --health-path) HEALTH_PATH="$2"; shift 2 ;;
+    --address) ADDRESS="$2"; shift 2 ;;
+    --data-root) DATA_ROOT="$2"; shift 2 ;;
+    --disks) DISKS="$2"; shift 2 ;;
+    --access-key) ACCESS_KEY="$2"; shift 2 ;;
+    --secret-key) SECRET_KEY="$2"; shift 2 ;;
+    --region) REGION="$2"; shift 2 ;;
+    --warp-bin) WARP_BIN="$2"; shift 2 ;;
+    --concurrency) CONCURRENCY="$2"; shift 2 ;;
+    --duration) DURATION="$2"; shift 2 ;;
+    --rounds) ROUNDS="$2"; shift 2 ;;
+    --cooldown) COOLDOWN_SECS="$2"; shift 2 ;;
+    --health-timeout) HEALTH_TIMEOUT_SECS="$2"; shift 2 ;;
+    --fail-pct) FAIL_PCT="$2"; shift 2 ;;
+    --warn-pct) WARN_PCT="$2"; shift 2 ;;
+    --allow-regression) ALLOW_REGRESSION=true; shift ;;
+    --exemption-reason) EXEMPTION_REASON="$2"; shift 2 ;;
+    --out-dir) OUT_DIR="$2"; shift 2 ;;
+    --dry-run) DRY_RUN=true; shift ;;
+    -h|--help) usage; exit 0 ;;
+    *) die "unknown argument: $1" ;;
+  esac
+done
+
+validate_positive_int "$DISKS" "--disks"
+validate_positive_int "$CONCURRENCY" "--concurrency"
+validate_positive_int "$ROUNDS" "--rounds"
+validate_positive_int "$COOLDOWN_SECS" "--cooldown"
+validate_positive_int "$HEALTH_TIMEOUT_SECS" "--health-timeout"
+validate_positive_int "$FAIL_PCT" "--fail-pct"
+validate_positive_int "$WARN_PCT" "--warn-pct"
+[[ "$ROUNDS" -ge 3 ]] || die "--rounds must be >= 3 for formal ABBA evidence"
+[[ -n "$BASELINE_BIN" ]] || die "--baseline-bin is required"
+[[ -n "$CANDIDATE_BIN" ]] || die "--candidate-bin is required"
+[[ "$DRY_RUN" == "true" || -x "$BASELINE_BIN" ]] || die "baseline binary is not executable: $BASELINE_BIN"
+[[ "$DRY_RUN" == "true" || -x "$CANDIDATE_BIN" ]] || die "candidate binary is not executable: $CANDIDATE_BIN"
+[[ -x "$ENHANCED_BENCH" ]] || die "missing load driver: $ENHANCED_BENCH"
+[[ -x "$GATE" ]] || die "missing gate: $GATE"
+if [[ "$DRY_RUN" != "true" ]] && ! command -v "$WARP_BIN" >/dev/null 2>&1; then
+  die "warp not found on PATH; install warp or pass --warp-bin"
+fi
+
+EXTERNAL=false
+if [[ -n "$ENDPOINT" ]]; then
+  EXTERNAL=true
+  ADDRESS="$ENDPOINT"
+  [[ -n "$DEPLOY_HOOK" ]] || log "warning: external mode without --deploy-hook; binaries must be swapped out of band"
+fi
+
+mkdir -p "$OUT_DIR"
+SERVER_LOG_DIR="$OUT_DIR/server-logs"
+run mkdir -p "$SERVER_LOG_DIR"
+
+SERVER_PID=""
+SERVER_LOG=""
+tear_down() {
+  [[ "$EXTERNAL" == "true" ]] && return 0
+  [[ -n "$SERVER_PID" ]] || return 0
+  run kill "$SERVER_PID" 2>/dev/null || true
+  SERVER_PID=""
+}
+trap tear_down EXIT INT TERM
+
+dump_server_log() {
+  [[ -n "$SERVER_LOG" && -f "$SERVER_LOG" ]] || return 0
+  echo "----- last 80 lines of $SERVER_LOG -----" >&2
+  tail -n 80 "$SERVER_LOG" >&2 || true
+  echo "----------------------------------------" >&2
+}
+
+wait_health() {
+  [[ "$DRY_RUN" == "true" ]] && return 0
+  local i
+  for ((i = 0; i < HEALTH_TIMEOUT_SECS; i++)); do
+    if [[ "$EXTERNAL" != "true" && -n "$SERVER_PID" ]] && ! kill -0 "$SERVER_PID" 2>/dev/null; then
+      echo "error: rustfs server (pid $SERVER_PID) exited before becoming healthy after ${i}s" >&2
+      dump_server_log
+      return 1
+    fi
+    if curl -fsS "http://${ADDRESS}${HEALTH_PATH}" >/dev/null 2>&1; then
+      return 0
+    fi
+    sleep 1
+  done
+  echo "error: endpoint http://${ADDRESS}${HEALTH_PATH} did not become healthy within ${HEALTH_TIMEOUT_SECS}s" >&2
+  dump_server_log
+  return 1
+}
+
+binary_for_leg() {
+  case "$1" in
+    A1|A2) echo "$BASELINE_BIN" ;;
+    B1|B2) echo "$CANDIDATE_BIN" ;;
+    *) die "unknown ABBA leg: $1" ;;
+  esac
+}
+
+phase_for_leg() {
+  case "$1" in
+    A1|A2) echo "baseline" ;;
+    B1|B2) echo "candidate" ;;
+    *) die "unknown ABBA leg: $1" ;;
+  esac
+}
+
+bring_up() {
+  local leg="$1" drive_sync="$2"
+  local phase bin
+  phase="$(phase_for_leg "$leg")"
+  bin="$(binary_for_leg "$leg")"
+
+  if [[ "$EXTERNAL" == "true" ]]; then
+    if [[ -n "$DEPLOY_HOOK" ]]; then
+      log "deploy hook: leg=$leg phase=$phase drive_sync=$drive_sync"
+      HOTPATH_ABBA_LEG="$leg" HOTPATH_ABBA_PHASE="$phase" HOTPATH_ABBA_BINARY="$bin" HOTPATH_ABBA_DRIVE_SYNC="$drive_sync" \
+        run bash -c "$DEPLOY_HOOK"
+    fi
+    wait_health
+    return 0
+  fi
+
+  local node_dir="$DATA_ROOT/$leg-sync-$drive_sync"
+  local disks=() d
+  for ((d = 1; d <= DISKS; d++)); do
+    disks+=("$node_dir/d$d")
+  done
+  run mkdir -p "${disks[@]}"
+  SERVER_LOG="$SERVER_LOG_DIR/$leg-sync-$drive_sync.log"
+
+  if [[ "$DRY_RUN" == "true" ]]; then
+    { printf 'DRY-RUN: RUSTFS_DRIVE_SYNC_ENABLE=%s %q server' "$drive_sync" "$bin"
+      printf ' %q' "${disks[@]}"; printf '\n'; } >&2
+    SERVER_PID="dry-run"
+    return 0
+  fi
+
+  cat >"$SERVER_LOG_DIR/$leg-sync-$drive_sync.env" <<EOF
+leg=$leg
+phase=$phase
+drive_sync=$drive_sync
+binary=$bin
+address=$ADDRESS
+disks=${disks[*]}
+health_url=http://${ADDRESS}${HEALTH_PATH}
+health_timeout_secs=$HEALTH_TIMEOUT_SECS
+uname=$(uname -a 2>/dev/null || echo unknown)
+warp_version=$("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
+EOF
+  RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true \
+  RUSTFS_ADDRESS="$ADDRESS" \
+  RUSTFS_ACCESS_KEY="$ACCESS_KEY" \
+  RUSTFS_SECRET_KEY="$SECRET_KEY" \
+  RUSTFS_REGION="$REGION" \
+  RUSTFS_CONSOLE_ENABLE=false \
+  RUSTFS_DRIVE_SYNC_ENABLE="$drive_sync" \
+    "$bin" server "${disks[@]}" >"$SERVER_LOG" 2>&1 &
+  SERVER_PID=$!
+  wait_health
+}
+
+measure() {
+  local leg="$1" workload="$2" mode="$3" size="$4" sync_label="$5" baseline_csv="${6:-}"
+  local cell="$OUT_DIR/$workload/$sync_label/$leg"
+  local args=(
+    --tool warp --warp-bin "$WARP_BIN" --warp-mode "$mode"
+    --endpoint "$ADDRESS" --access-key "$ACCESS_KEY" --secret-key "$SECRET_KEY"
+    --region "$REGION" --sizes "$size" --concurrency "$CONCURRENCY"
+    --duration "$DURATION" --rounds "$ROUNDS" --cooldown-secs "$COOLDOWN_SECS"
+    --out-dir "$cell"
+  )
+  [[ -n "$baseline_csv" ]] && args+=(--baseline-csv "$baseline_csv")
+  run "$ENHANCED_BENCH" "${args[@]}" >&2
+  echo "$cell"
+}
+
+write_schedule_header() {
+  echo "sync_label,drive_sync,workload,mode,size,leg,phase,binary,out_dir" >"$OUT_DIR/abba_schedule.csv"
+}
+
+append_schedule() {
+  local sync_label="$1" drive_sync="$2" workload="$3" mode="$4" size="$5" leg="$6"
+  local phase bin
+  phase="$(phase_for_leg "$leg")"
+  bin="$(binary_for_leg "$leg")"
+  echo "$sync_label,$drive_sync,$workload,$mode,$size,$leg,$phase,$bin,$OUT_DIR/$workload/$sync_label/$leg" >>"$OUT_DIR/abba_schedule.csv"
+}
+
+write_manifest() {
+  cat >"$OUT_DIR/manifest.env" <<EOF
+generated_at_utc=$(date -u +%Y-%m-%dT%H:%M:%SZ 2>/dev/null || echo unknown)
+runner=$(uname -srm 2>/dev/null || echo unknown)
+schedule=ABBA
+rounds=$ROUNDS
+duration=$DURATION
+cooldown_secs=$COOLDOWN_SECS
+concurrency=$CONCURRENCY
+baseline_bin=$BASELINE_BIN
+candidate_bin=$CANDIDATE_BIN
+external=$EXTERNAL
+endpoint=$ADDRESS
+warp_version=$("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
+EOF
+}
+
+declare -a CANDIDATE_COMPARE_CSVS=()
+declare -a DRIFT_COMPARE_CSVS=()
+
+write_manifest
+write_schedule_header
+
+for ds_spec in "${DRIVE_SYNC_MATRIX[@]}"; do
+  IFS='|' read -r sync_label drive_sync <<<"$ds_spec"
+
+  for leg in A1 B1 B2 A2; do
+    log "=== $sync_label leg $leg ($(phase_for_leg "$leg")) ==="
+    bring_up "$leg" "$drive_sync"
+
+    for wl_spec in "${WORKLOADS[@]}"; do
+      IFS='|' read -r workload mode size <<<"$wl_spec"
+      append_schedule "$sync_label" "$drive_sync" "$workload" "$mode" "$size" "$leg"
+
+      baseline_csv=""
+      if [[ "$leg" != "A1" ]]; then
+        baseline_csv="$OUT_DIR/$workload/$sync_label/A1/median_summary.csv"
+      fi
+
+      cell="$(measure "$leg" "$workload" "$mode" "$size" "$sync_label" "$baseline_csv")"
+      case "$leg" in
+        B1|B2) CANDIDATE_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
+        A2) DRIFT_COMPARE_CSVS+=("$cell/baseline_compare.csv") ;;
+      esac
+    done
+
+    tear_down
+  done
+done
+
+gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/candidate_gate.md")
+for csv in "${CANDIDATE_COMPARE_CSVS[@]}"; do
+  gate_args+=(--compare-csv "$csv")
+done
+[[ "$ALLOW_REGRESSION" == "true" ]] && gate_args+=(--allow-regression --exemption-reason "$EXEMPTION_REASON")
+
+drift_gate_args=(--fail-pct "$FAIL_PCT" --warn-pct "$WARN_PCT" --markdown "$OUT_DIR/baseline_drift_gate.md")
+for csv in "${DRIFT_COMPARE_CSVS[@]}"; do
+  drift_gate_args+=(--compare-csv "$csv")
+done
+
+if [[ "$DRY_RUN" == "true" ]]; then
+  log "dry-run complete; candidate compare CSVs=${#CANDIDATE_COMPARE_CSVS[@]} baseline drift CSVs=${#DRIFT_COMPARE_CSVS[@]}"
+  { printf 'DRY-RUN:'; printf ' %q' "$GATE" "${gate_args[@]}"; printf '\n'; } >&2
+  { printf 'DRY-RUN:'; printf ' %q' "$GATE" "${drift_gate_args[@]}"; printf '\n'; } >&2
+  exit 0
+fi
+
+log "applying candidate relative-budget gate"
+set +e
+"$GATE" "${gate_args[@]}"
+candidate_status=$?
+set -e
+
+log "applying A2-vs-A1 baseline drift gate"
+set +e
+"$GATE" "${drift_gate_args[@]}"
+drift_status=$?
+set -e
+
+cat >"$OUT_DIR/summary.md" <<EOF
+# Hotpath Warp ABBA Summary
+
+- schedule: A1 baseline -> B1 candidate -> B2 candidate -> A2 baseline
+- runner: $(uname -srm 2>/dev/null || echo unknown)
+- warp: $("$WARP_BIN" --version 2>/dev/null | head -n1 || echo unknown)
+- matrix: duration=$DURATION rounds=$ROUNDS cooldown=$COOLDOWN_SECS disks=$DISKS concurrency=$CONCURRENCY
+- endpoint: $ADDRESS
+- baseline binary: $BASELINE_BIN
+- candidate binary: $CANDIDATE_BIN
+- candidate gate: $OUT_DIR/candidate_gate.md (exit $candidate_status)
+- baseline drift gate: $OUT_DIR/baseline_drift_gate.md (exit $drift_status)
+
+Interpretation:
+
+- Treat candidate gate failures as actionable only when the A2-vs-A1 drift gate is PASS or the affected workload's A2 drift is materially smaller than the B1/B2 candidate delta.
+- If both candidate and baseline drift fail on the same workload, rerun with longer duration, more rounds, or a quieter runner before assigning causality.
+EOF
+
+log "summary written to $OUT_DIR/summary.md"
+if [[ "$candidate_status" -ne 0 || "$drift_status" -ne 0 ]]; then
+  exit 1
+fi