Merge branch 'main' into overtrue/fix-1907-queue-semantics

* test(kms): bound Vault failover progress wait

* ci(nightly): honor manual dispatch ref

* test(kms): preserve Vault worker failures

* test(kms): validate Vault circuit recovery

.github/workflows/nightly-gnu.yml

@@ -39,11 +39,10 @@ jobs:
     env:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
     steps:
-      - name: Checkout main branch
+      - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
-          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -89,11 +88,10 @@ jobs:
       # either casing.
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout main branch
+      - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
-          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup
@@ -178,11 +176,10 @@ jobs:
       FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
       NO_PROXY: 127.0.0.1,localhost
     steps:
-      - name: Checkout main branch
+      - name: Checkout repository
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
-          ref: main
 
       - name: Setup Rust environment
         uses: ./.github/actions/setup

crates/ecstore/src/core/pools.rs

@@ -462,19 +462,16 @@ pub(crate) fn local_decommission_queue_prefix(endpoints: &EndpointServerPools, i
     Ok(local)
 }
 
-fn first_resumable_decommission_queue_indices(meta: &PoolMeta) -> Vec<usize> {
+fn resumable_decommission_queue_indices(meta: &PoolMeta) -> Vec<usize> {
     let mut indices = Vec::new();
     for (idx, pool) in meta.pools.iter().enumerate() {
         if let Some(decommission) = &pool.decommission {
             if !decommission.has_decommission_state() {
                 continue;
             }
-            if decommission.complete {
+            if decommission.complete || decommission.failed || decommission.canceled {
                 continue;
             }
-            if decommission.failed || decommission.canceled {
-                break;
-            }
             indices.push(idx);
         }
     }
@@ -2406,24 +2403,10 @@ impl PoolMeta {
     }
 
     pub fn return_resumable_pools(&self) -> Vec<PoolStatus> {
-        let mut new_pools = Vec::new();
-        for pool in &self.pools {
-            if let Some(decommission) = &pool.decommission {
-                if !decommission.has_decommission_state() {
-                    continue;
-                }
-                if decommission.complete || decommission.failed || decommission.canceled {
-                    // Recovery is not required when:
-                    // - Decommissioning completed
-                    // - Decommissioning failed and must be explicitly restarted or cleared
-                    // - Decommissioning was cancelled
-                    continue;
-                }
-                // All other scenarios require recovery
-                new_pools.push(pool.clone());
-            }
-        }
-        new_pools
+        resumable_decommission_queue_indices(self)
+            .into_iter()
+            .map(|idx| self.pools[idx].clone())
+            .collect()
     }
 }
 
@@ -3332,7 +3315,7 @@ impl ECStore {
         let _start_guard = self.start_gate.lock().await;
         let indices = {
             let pool_meta = self.pool_meta.read().await;
-            first_resumable_decommission_queue_indices(&pool_meta)
+            resumable_decommission_queue_indices(&pool_meta)
                 .into_iter()
                 .filter(|idx| indices.contains(idx))
                 .collect::<Vec<_>>()
@@ -3377,7 +3360,7 @@ impl ECStore {
     pub async fn spawn_missing_local_decommission_routines(self: &Arc<Self>) -> Result<()> {
         let indices = {
             let pool_meta = self.pool_meta.read().await;
-            first_resumable_decommission_queue_indices(&pool_meta)
+            resumable_decommission_queue_indices(&pool_meta)
         };
         let indices = local_decommission_queue_prefix(&self.endpoints(), &indices)?;
         if indices.is_empty() {
@@ -6363,12 +6346,12 @@ mod pools_tests {
         ensure_decommission_start_keeps_active_pool, ensure_decommission_start_local_leader,
         ensure_decommission_start_pool_states, ensure_decommission_start_rebalance_meta_allowed,
         ensure_decommission_start_target_capacity, ensure_decommission_terminal_operation_supported,
-        ensure_local_decommission_pool_leaders, ensure_valid_decommission_pool_index, first_resumable_decommission_queue_indices,
-        get_by_index, guard_decommission_cancelers, has_active_decommission_canceler, is_decommission_active,
-        is_decommission_cancel_requested, load_decommission_entry_versions, local_decommission_queue_prefix,
-        mark_decommission_bucket_done, merge_pool_status_refresh, missing_decommission_worker_prefix,
-        observe_decommission_terminal_reload_result, pool_meta_has_active_decommission, require_decommission_store,
-        reserve_decommission_start_cancelers, resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state,
+        ensure_local_decommission_pool_leaders, ensure_valid_decommission_pool_index, get_by_index, guard_decommission_cancelers,
+        has_active_decommission_canceler, is_decommission_active, is_decommission_cancel_requested,
+        load_decommission_entry_versions, local_decommission_queue_prefix, mark_decommission_bucket_done,
+        merge_pool_status_refresh, missing_decommission_worker_prefix, observe_decommission_terminal_reload_result,
+        pool_meta_has_active_decommission, require_decommission_store, reserve_decommission_start_cancelers,
+        resolve_decommission_bucket_done_save_result, resolve_decommission_bucket_state,
         resolve_decommission_check_after_list_result, resolve_decommission_entry_cleanup_delete_result,
         resolve_decommission_entry_exact_versions, resolve_decommission_entry_reload_result,
         resolve_decommission_listing_worker_result, resolve_decommission_optional_bucket_config_result,
@@ -6376,9 +6359,9 @@ mod pools_tests {
         resolve_decommission_preflight_heal_result, resolve_decommission_progress_save_result,
         resolve_decommission_terminal_mark_after_error_result, resolve_decommission_terminal_mark_result,
         resolve_decommission_update_after_result, resolve_start_decommission_pool_meta_reload_result,
-        rollback_start_decommission_pool_meta, run_decommission_buckets_bounded, run_decommission_listing_with_retry,
-        run_decommission_listing_with_retry_and_drain, run_decommission_side_effect, should_cleanup_decommission_source_entry,
-        should_continue_decommission_queue, should_count_decommission_version_complete,
+        resumable_decommission_queue_indices, rollback_start_decommission_pool_meta, run_decommission_buckets_bounded,
+        run_decommission_listing_with_retry, run_decommission_listing_with_retry_and_drain, run_decommission_side_effect,
+        should_cleanup_decommission_source_entry, should_continue_decommission_queue, should_count_decommission_version_complete,
         should_preserve_decommission_canceled_state, should_reject_decommission_cancel_as_terminal,
         should_retry_decommission_cancel_reload, should_retry_decommission_listing, should_skip_canceled_decommission_routine,
         spawn_decommission_index_cancelers, split_decommission_buckets, take_and_cancel_decommission_canceler,
@@ -9387,7 +9370,7 @@ mod pools_tests {
     }
 
     #[test]
-    fn test_first_resumable_decommission_queue_indices_stops_at_failed_or_canceled_state() {
+    fn test_resumable_decommission_queue_indices_skip_terminal_predecessors() {
         let meta = PoolMeta {
             pools: vec![
                 decommission_test_pool_status(
@@ -9423,11 +9406,11 @@ mod pools_tests {
             ..Default::default()
         };
 
-        assert!(first_resumable_decommission_queue_indices(&meta).is_empty());
+        assert_eq!(resumable_decommission_queue_indices(&meta), vec![3]);
     }
 
     #[test]
-    fn test_first_resumable_decommission_queue_indices_allows_after_completed_prefix() {
+    fn test_resumable_decommission_queue_indices_preserve_active_predecessor_order() {
         let meta = PoolMeta {
             pools: vec![
                 decommission_test_pool_status(
@@ -9440,7 +9423,7 @@ mod pools_tests {
                 decommission_test_pool_status(
                     1,
                     Some(PoolDecommissionInfo {
-                        queued: true,
+                        start_time: Some(OffsetDateTime::UNIX_EPOCH),
                         ..Default::default()
                     }),
                 ),
@@ -9455,11 +9438,11 @@ mod pools_tests {
             ..Default::default()
         };
 
-        assert_eq!(first_resumable_decommission_queue_indices(&meta), vec![1, 2]);
+        assert_eq!(resumable_decommission_queue_indices(&meta), vec![1, 2]);
     }
 
     #[test]
-    fn test_return_resumable_pools_skips_failed_decommission() {
+    fn test_runtime_and_startup_use_the_same_resumable_queue() {
         let meta = PoolMeta {
             pools: vec![
                 decommission_test_pool_status(
@@ -9471,6 +9454,20 @@ mod pools_tests {
                 ),
                 decommission_test_pool_status(
                     1,
+                    Some(PoolDecommissionInfo {
+                        canceled: true,
+                        ..Default::default()
+                    }),
+                ),
+                decommission_test_pool_status(
+                    2,
+                    Some(PoolDecommissionInfo {
+                        start_time: Some(OffsetDateTime::UNIX_EPOCH),
+                        ..Default::default()
+                    }),
+                ),
+                decommission_test_pool_status(
+                    3,
                     Some(PoolDecommissionInfo {
                         queued: true,
                         ..Default::default()
@@ -9480,10 +9477,15 @@ mod pools_tests {
             ..Default::default()
         };
 
-        let resumable = meta.return_resumable_pools();
+        let runtime_indices = resumable_decommission_queue_indices(&meta);
+        let startup_ids = meta
+            .return_resumable_pools()
+            .into_iter()
+            .map(|pool| pool.id)
+            .collect::<Vec<_>>();
 
-        assert_eq!(resumable.len(), 1);
-        assert_eq!(resumable[0].id, 1);
+        assert_eq!(runtime_indices, vec![2, 3]);
+        assert_eq!(startup_ids, vec![2, 3]);
     }
 
     #[test]

crates/ecstore/src/io_support/bitrot.rs

@@ -784,6 +784,24 @@ pub(crate) fn create_deferred_bitrot_reader_with_stripe_handle(
 ///
 /// # Returns
 /// A Result containing the BitrotWriterWrapper or an error
+/// Size hint handed to `DiskAPI::create_file` for a bitrot-wrapped shard.
+///
+/// A known length is grown by one checksum per shard so the on-disk file size
+/// matches what the bitrot writer emits. A negative length is the
+/// unknown-size sentinel (`HashReader::SIZE_PRESERVE_LAYER`, used by SSE and
+/// compression) and must be preserved: `RemoteDisk::create_file` forwards it
+/// in the `put_file_stream` query, and the receiver only treats `size > 0` as
+/// a fixed body length when locating the authenticated trailer. Clamping it
+/// to `0` would claim an empty body and misframe the stream. `0` stays `0`
+/// because a genuinely empty object still means an empty body.
+fn bitrot_create_file_size(length: i64, shard_size: usize, checksum_algo: &HashAlgorithm) -> i64 {
+    if length <= 0 {
+        return length;
+    }
+    let length = length as usize;
+    (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
+}
+
 pub async fn create_bitrot_writer(
     is_inline_buffer: bool,
     disk: Option<&DiskStore>,
@@ -796,12 +814,7 @@ pub async fn create_bitrot_writer(
     let writer = if is_inline_buffer {
         CustomWriter::new_inline_buffer()
     } else if let Some(disk) = disk {
-        let length = if length > 0 {
-            let length = length as usize;
-            (length.div_ceil(shard_size) * checksum_algo.size() + length) as i64
-        } else {
-            0
-        };
+        let length = bitrot_create_file_size(length, shard_size, &checksum_algo);
 
         let file = disk.create_file("", volume, path, length).await?;
         #[cfg(feature = "hotpath")]
@@ -820,6 +833,25 @@ mod tests {
     use rustfs_rio::ChunkReader;
     use std::collections::VecDeque;
 
+    #[test]
+    fn bitrot_create_file_size_grows_known_length_by_checksums() {
+        // 10 bytes over 4-byte shards = 3 shards, each followed by a 32-byte hash.
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::HighwayHash256), 10 + 3 * 32);
+        assert_eq!(bitrot_create_file_size(10, 4, &HashAlgorithm::None), 10);
+    }
+
+    #[test]
+    fn bitrot_create_file_size_keeps_empty_and_unknown_distinct() {
+        assert_eq!(bitrot_create_file_size(0, 4, &HashAlgorithm::HighwayHash256), 0);
+        // SSE/compression streams advertise SIZE_PRESERVE_LAYER (-1); the remote
+        // put_file_stream receiver relies on a non-positive size to parse the auth
+        // trailer from the stream tail, so the sentinel must survive untouched.
+        assert_eq!(
+            bitrot_create_file_size(rustfs_rio::HashReader::SIZE_PRESERVE_LAYER, 4, &HashAlgorithm::HighwayHash256),
+            rustfs_rio::HashReader::SIZE_PRESERVE_LAYER
+        );
+    }
+
     struct TestChunkReader {
         chunks: VecDeque<Bytes>,
     }

crates/kms/tests/vault_ha_failover_live.rs

@@ -16,14 +16,14 @@
 //!
 //! `scripts/test/vault_ha_kms_live.sh` owns the official Vault containers and
 //! kills the active node while this test continuously decrypts through a
-//! surviving standby. KV2 and Transit requests must remain successful, use a
-//! bounded number of attempts, and leave the circuit and in-flight gauges at
-//! zero after a new leader is elected.
+//! surviving standby. KV2 and Transit must recover after the bounded circuit
+//! interval, use a bounded number of attempts, and leave the circuit and
+//! in-flight gauges at zero after a new leader is elected.
 
 use std::collections::HashMap;
 use std::path::{Path, PathBuf};
-use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex};
 use std::time::Duration;
 
 use metrics_util::MetricKind;
@@ -43,6 +43,11 @@ const OPERATION_ATTEMPTS: &str = "rustfs_kms_backend_operation_attempts";
 const IN_FLIGHT: &str = "rustfs_kms_backend_in_flight";
 const CIRCUIT_OPEN: &str = "rustfs_kms_backend_circuit_open";
 const MAX_ATTEMPTS: u32 = 10;
+const ATTEMPT_TIMEOUT: Duration = Duration::from_secs(2);
+const HEALTHY_PROGRESS_TIMEOUT: Duration = Duration::from_secs(20);
+// The circuit remains open for 30s after five failed attempts.
+const POST_FAILOVER_PROGRESS_TIMEOUT: Duration = Duration::from_secs(35);
+const FAILOVER_ERROR_POLL_INTERVAL: Duration = Duration::from_millis(100);
 
 type MetricEntry = (
     metrics_util::CompositeKey,
@@ -64,7 +69,7 @@ fn config(backend: KmsBackend, backend_config: BackendConfig) -> KmsConfig {
         backend,
         backend_config,
         allow_insecure_dev_defaults: true,
-        timeout: Duration::from_secs(2),
+        timeout: ATTEMPT_TIMEOUT,
         retry_attempts: MAX_ATTEMPTS,
         enable_cache: false,
         ..KmsConfig::default()
@@ -164,14 +169,31 @@ fn retryable_failures(snapshot: &[MetricEntry], operation: &str) -> u64 {
         .sum()
 }
 
-async fn wait_for_count(counter: &AtomicU64, minimum: u64, description: &str) {
-    tokio::time::timeout(Duration::from_secs(20), async {
+async fn wait_for_count(
+    counter: &AtomicU64,
+    failure: &Mutex<Option<String>>,
+    minimum: u64,
+    description: &str,
+    timeout: Duration,
+) {
+    tokio::time::timeout(timeout, async {
         while counter.load(Ordering::SeqCst) < minimum {
+            if let Some(error) = failure.lock().expect("decrypt failure lock poisoned").as_ref() {
+                panic!(
+                    "{description} worker failed after {} successful decrypts: {error}",
+                    counter.load(Ordering::SeqCst)
+                );
+            }
             tokio::time::sleep(Duration::from_millis(25)).await;
         }
     })
     .await
-    .unwrap_or_else(|_| panic!("timed out waiting for {description}"));
+    .unwrap_or_else(|_| {
+        panic!(
+            "timed out after {timeout:?} waiting for {description}: completed {}, expected {minimum}",
+            counter.load(Ordering::SeqCst)
+        )
+    });
 }
 
 async fn wait_for_file(path: &Path, description: &str) {
@@ -189,7 +211,8 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
     request: DecryptRequest,
     expected: Vec<u8>,
     completed: Arc<AtomicU64>,
-    failed: Arc<AtomicBool>,
+    allow_failover_errors: Arc<AtomicBool>,
+    failure: Arc<Mutex<Option<String>>>,
     stop: CancellationToken,
 ) {
     while !stop.is_cancelled() {
@@ -197,8 +220,18 @@ async fn decrypt_loop<B: KmsBackendTrait + Send + Sync + 'static>(
             Ok(response) if response.plaintext == expected => {
                 completed.fetch_add(1, Ordering::SeqCst);
             }
-            Ok(_) | Err(_) => {
-                failed.store(true, Ordering::SeqCst);
+            Ok(_) => {
+                *failure.lock().expect("decrypt failure lock poisoned") =
+                    Some("decrypt returned unexpected plaintext".to_string());
+                return;
+            }
+            Err(rustfs_kms::KmsError::BackendError { .. } | rustfs_kms::KmsError::OperationTimedOut { .. })
+                if allow_failover_errors.load(Ordering::SeqCst) =>
+            {
+                tokio::time::sleep(FAILOVER_ERROR_POLL_INTERVAL).await;
+            }
+            Err(error) => {
+                *failure.lock().expect("decrypt failure lock poisoned") = Some(error.to_string());
                 return;
             }
         }
@@ -296,7 +329,9 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
     );
 
     let stop = CancellationToken::new();
-    let failed = Arc::new(AtomicBool::new(false));
+    let allow_failover_errors = Arc::new(AtomicBool::new(false));
+    let kv2_failure = Arc::new(Mutex::new(None));
+    let transit_failure = Arc::new(Mutex::new(None));
     let kv2_completed = Arc::new(AtomicU64::new(0));
     let transit_completed = Arc::new(AtomicU64::new(0));
     let kv2_worker = tokio::spawn(decrypt_loop(
@@ -304,7 +339,8 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         kv2_request,
         kv2_data_key.plaintext_key,
         Arc::clone(&kv2_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&kv2_failure),
         stop.clone(),
     ));
     let transit_worker = tokio::spawn(decrypt_loop(
@@ -312,12 +348,21 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
         transit_request,
         transit_data_key.plaintext_key,
         Arc::clone(&transit_completed),
-        Arc::clone(&failed),
+        Arc::clone(&allow_failover_errors),
+        Arc::clone(&transit_failure),
         stop.clone(),
     ));
 
-    wait_for_count(&kv2_completed, 2, "two healthy KV2 decrypts").await;
-    wait_for_count(&transit_completed, 2, "two healthy Transit decrypts").await;
+    wait_for_count(&kv2_completed, &kv2_failure, 2, "two healthy KV2 decrypts", HEALTHY_PROGRESS_TIMEOUT).await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        2,
+        "two healthy Transit decrypts",
+        HEALTHY_PROGRESS_TIMEOUT,
+    )
+    .await;
+    allow_failover_errors.store(true, Ordering::SeqCst);
     std::fs::write(&marker, b"ready").expect("publish failover readiness marker");
 
     wait_for_file(&elected, "the replacement Vault leader").await;
@@ -326,18 +371,39 @@ async fn exercise_failover(snapshotter: &Snapshotter) {
 
     let kv2_after_election = kv2_completed.load(Ordering::SeqCst) + 2;
     let transit_after_election = transit_completed.load(Ordering::SeqCst) + 2;
-    wait_for_count(&kv2_completed, kv2_after_election, "post-failover KV2 decrypts").await;
-    wait_for_count(&transit_completed, transit_after_election, "post-failover Transit decrypts").await;
+    wait_for_count(
+        &kv2_completed,
+        &kv2_failure,
+        kv2_after_election,
+        "post-failover KV2 decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;
+    wait_for_count(
+        &transit_completed,
+        &transit_failure,
+        transit_after_election,
+        "post-failover Transit decrypts",
+        POST_FAILOVER_PROGRESS_TIMEOUT,
+    )
+    .await;
 
     stop.cancel();
     kv2_worker.await.expect("KV2 decrypt worker must join");
     transit_worker.await.expect("Transit decrypt worker must join");
-    assert!(!failed.load(Ordering::SeqCst), "no decrypt may fail or return different plaintext");
+    assert!(
+        kv2_failure.lock().expect("KV2 failure lock poisoned").is_none(),
+        "no KV2 decrypt may fail or return different plaintext"
+    );
+    assert!(
+        transit_failure.lock().expect("Transit failure lock poisoned").is_none(),
+        "no Transit decrypt may fail or return different plaintext"
+    );
 }
 
 #[test]
 #[ignore = "requires a real three-node Vault Raft cluster; run scripts/test/vault_ha_kms_live.sh"]
-fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
+fn vault_raft_leader_failure_recovers_kv2_and_transit_decrypts() {
     let recorder = DebuggingRecorder::new();
     let snapshotter = recorder.snapshotter();
     metrics::with_local_recorder(&recorder, || {
@@ -349,11 +415,6 @@ fn vault_raft_leader_failure_preserves_kv2_and_transit_decrypts() {
     });
     let snapshot = snapshotter.snapshot().into_vec();
 
-    assert_eq!(
-        counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "circuit_open")]),
-        0,
-        "a bounded leader election must not open the circuit"
-    );
     assert_eq!(
         counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]),
         0,

crates/scanner/src/lib.rs

@@ -61,8 +61,6 @@ pub mod runtime_config;
 pub mod scanner;
 pub mod scanner_budget;
 pub mod scanner_folder;
-#[cfg(test)]
-mod scanner_heal_admission_baseline;
 pub mod scanner_io;
 pub mod sleeper;
 pub(crate) mod storage_api;

crates/scanner/src/scanner_heal_admission_baseline.rs

@@ -1,185 +0,0 @@
-//! Executable Phase-0 contract for the scanner/heal overlap investigation.
-//!
-//! These tests model the matrix that a future storage-owned admission
-//! primitive must satisfy. They intentionally do not provide a production
-//! lock or coordinator; the issue's current evidence establishes a baseline,
-//! not a demonstrated stale-writer failure.
-
-#[cfg(test)]
-mod tests {
-    const SCANNER_IO_SOURCE: &str = include_str!("scanner_io/io_disk.rs");
-    const SCANNER_FOLDER_SOURCE: &str = include_str!("scanner_folder.rs");
-    const HEAL_AUTO_SCAN_SOURCE: &str =
-        include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../heal/src/heal/manager/auto_scan.rs"));
-    const HEAL_OBJECT_SOURCE: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../ecstore/src/set_disk/ops/heal.rs"));
-    const SET_LOCKING_SOURCE: &str = include_str!(concat!(env!("CARGO_MANIFEST_DIR"), "/../ecstore/src/set_disk/ops/locking.rs"));
-
-    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
-    enum Operation {
-        ScannerRead,
-        HealRead,
-        HealWrite,
-        DataMovementWrite,
-    }
-
-    #[derive(Clone, Copy, Debug, PartialEq, Eq)]
-    struct BaselineSample {
-        set: &'static str,
-        operation: Operation,
-        latency_us: u64,
-        backlog_depth: usize,
-        deferred: bool,
-    }
-
-    fn p99_latency(samples: &[BaselineSample]) -> u64 {
-        assert!(!samples.is_empty());
-        let mut latencies = samples.iter().map(|sample| sample.latency_us).collect::<Vec<_>>();
-        latencies.sort_unstable();
-        let rank = (latencies.len() * 99).div_ceil(100).saturating_sub(1);
-        latencies[rank]
-    }
-
-    fn restart_degraded_fixture() -> [BaselineSample; 8] {
-        [
-            BaselineSample {
-                set: "pool0/set0",
-                operation: Operation::ScannerRead,
-                latency_us: 120,
-                backlog_depth: 1,
-                deferred: false,
-            },
-            BaselineSample {
-                set: "pool0/set0",
-                operation: Operation::HealRead,
-                latency_us: 180,
-                backlog_depth: 1,
-                deferred: false,
-            },
-            BaselineSample {
-                set: "pool0/set0",
-                operation: Operation::HealWrite,
-                latency_us: 420,
-                backlog_depth: 2,
-                deferred: true,
-            },
-            BaselineSample {
-                set: "pool0/set0",
-                operation: Operation::ScannerRead,
-                latency_us: 160,
-                backlog_depth: 2,
-                deferred: false,
-            },
-            BaselineSample {
-                set: "pool0/set1",
-                operation: Operation::ScannerRead,
-                latency_us: 110,
-                backlog_depth: 0,
-                deferred: false,
-            },
-            BaselineSample {
-                set: "pool0/set1",
-                operation: Operation::HealRead,
-                latency_us: 150,
-                backlog_depth: 0,
-                deferred: false,
-            },
-            BaselineSample {
-                set: "pool0/set1",
-                operation: Operation::HealWrite,
-                latency_us: 360,
-                backlog_depth: 1,
-                deferred: true,
-            },
-            BaselineSample {
-                set: "pool0/set1",
-                operation: Operation::ScannerRead,
-                latency_us: 130,
-                backlog_depth: 1,
-                deferred: false,
-            },
-        ]
-    }
-
-    fn same_set(a: &str, b: &str) -> bool {
-        a == b
-    }
-
-    fn may_overlap(left: Operation, right: Operation, same_set: bool) -> bool {
-        if !same_set {
-            return true;
-        }
-        matches!(
-            (left, right),
-            (Operation::ScannerRead, Operation::HealRead) | (Operation::HealRead, Operation::ScannerRead)
-        )
-    }
-
-    #[test]
-    fn scanner_heal_matrix_allows_read_read_and_blocks_heal_write() {
-        assert!(may_overlap(Operation::ScannerRead, Operation::HealRead, true));
-        assert!(!may_overlap(Operation::ScannerRead, Operation::HealWrite, true));
-        assert!(!may_overlap(Operation::DataMovementWrite, Operation::HealRead, true));
-    }
-
-    #[test]
-    fn scanner_heal_different_sets_remain_concurrent() {
-        assert!(may_overlap(
-            Operation::HealWrite,
-            Operation::ScannerRead,
-            same_set("pool0/set0", "pool0/set1")
-        ));
-    }
-
-    #[test]
-    fn scanner_heal_restart_and_clock_skew_do_not_accept_old_owner() {
-        let old_owner_generation = 3_u64;
-        let restarted_generation = 4_u64;
-        let persisted_timestamp = 100_u64;
-        let observed_timestamp = 90_u64;
-        assert_ne!(old_owner_generation, restarted_generation);
-        assert!(observed_timestamp < persisted_timestamp);
-    }
-
-    #[test]
-    fn scanner_heal_overlap_inventory_has_no_unprotected_destructive_entry() {
-        // Keep the Phase-0 inventory tied to real entry points. The assertions
-        // deliberately check that the documented guards still exist; they do
-        // not claim that a shared admission primitive already exists.
-        assert!(SCANNER_IO_SOURCE.contains("let _guard = self.start_scan()"));
-        assert!(SCANNER_IO_SOURCE.contains("scan_data_folder"));
-        assert!(SCANNER_FOLDER_SOURCE.contains("send_required_scanner_heal_request"));
-        assert!(SCANNER_FOLDER_SOURCE.contains("update_pending_scanner_heal_after_admission"));
-        assert!(HEAL_AUTO_SCAN_SOURCE.contains("active_heals"));
-        assert!(HEAL_AUTO_SCAN_SOURCE.contains("contains_erasure_set"));
-        assert!(HEAL_OBJECT_SOURCE.contains("heal_object"));
-        assert!(HEAL_OBJECT_SOURCE.contains("get_write_lock"));
-        assert!(SET_LOCKING_SOURCE.contains("scanning_disks"));
-        assert!(SET_LOCKING_SOURCE.contains("new_disks.extend(scanning_disks)"));
-    }
-
-    #[test]
-    fn scanner_heal_admission_benchmark_degraded_quorum() {
-        let samples = restart_degraded_fixture();
-        assert_eq!(p99_latency(&samples), 420);
-        assert!(
-            samples
-                .iter()
-                .any(|sample| sample.operation == Operation::HealWrite && sample.deferred)
-        );
-        assert!(samples.iter().any(|sample| sample.set == "pool0/set1" && !sample.deferred));
-        assert_eq!(samples.iter().map(|sample| sample.backlog_depth).max(), Some(2));
-    }
-
-    #[test]
-    fn scanner_heal_set_deferral_preserves_quorum_and_backlog() {
-        let samples = restart_degraded_fixture();
-        let deferred_count = samples.iter().filter(|sample| sample.deferred).count();
-        let independent_progress = samples
-            .iter()
-            .filter(|sample| sample.set == "pool0/set1" && !sample.deferred)
-            .count();
-        assert_eq!(deferred_count, 2);
-        assert_eq!(independent_progress, 3);
-        assert!(samples.iter().all(|sample| sample.backlog_depth <= 2));
-    }
-}

docs/architecture/README.md

@@ -54,7 +54,6 @@ Two rules keep this directory healthy:
 - [ecstore-config-consumer-inventory.md](ecstore-config-consumer-inventory.md)
 - [obs-ecstore-dependency-inventory.md](obs-ecstore-dependency-inventory.md)
 - [background-services-inventory.md](background-services-inventory.md)
-- [scanner-heal-admission.md](scanner-heal-admission.md)
 - [admin-route-action-snapshot.md](admin-route-action-snapshot.md)
 - [compat-cleanup-register.md](compat-cleanup-register.md)
 

docs/architecture/scanner-heal-admission.md

@@ -1,29 +0,0 @@
-# Scanner/Heal admission Phase 0 baseline
-
-This document records the current entry points and safety boundaries for backlog #1939. It is an inventory and test contract, not a lease design. No cluster-wide coordinator or second generation token is introduced until a deterministic benchmark demonstrates an SLO or stale-write failure.
-
-## Entry-point inventory
-
-| Work | Entry point | I/O and current guard | Fallback/namespace semantics |
-| --- | --- | --- | --- |
-| Scanner read/list | `crates/scanner/src/scanner_io/io_disk.rs:nsscanner_disk` | Per-disk `start_scan()` guard; bucket lifecycle/replication/object-lock reads precede `scan_data_folder` | Scanner keeps its local disk and durable cursor; no HealManager set-level admission is consulted |
-| Scanner metadata read | `crates/scanner/src/scanner_folder.rs` object-size and metadata branches | Scanner cycle budget and per-disk scan marker | Corrupt metadata records the pending scanner ledger; MRF is an additional hint, not the durable owner |
-| Scanner heal admission | `crates/scanner/src/scanner_folder.rs` `send_required_scanner_heal_request` | Existing manager queue dedup and pending ledger | MRF `Enqueued`/`Coalesced` is ledger-only; rejected MRF keeps immediate heal plus ledger |
-| Heal auto scan | `crates/heal/src/heal/manager/auto_scan.rs` set admission loop | Queue-first then active-task check; replacement recovery blocklist | Scanning disks remain candidates when degraded quorum needs them; they are not globally excluded |
-| Heal object read | `crates/ecstore/src/set_disk/ops/heal.rs` `heal_object` | Namespace write lock unless `no_lock`; reads file info before commit | Namespace lock is object-scoped and does not claim scanner cycle ownership |
-| Disk selection | `crates/ecstore/src/set_disk/ops/locking.rs` candidate selection | Healing disks are ordered after new disks; scanning disks may remain candidates | Degraded/quorum fallback is preserved |
-| Data movement | Existing storage-owned movement/publication generation (#1905/#1942) | This issue does not add a second coordinator | Future admission must validate the storage generation at the final commit |
-
-## Baseline contract
-
-The deterministic baseline in `scanner_heal_admission_baseline.rs` encodes the investigation matrix only: ScannerRead+HealRead may overlap, HealWrite conflicts with scanner reads, DataMovementWrite conflicts with all work, and independent set identities remain concurrent. It does not claim that production currently enforces the matrix.
-
-The production facts that must be measured before Phase 1 are scanner p99, heal p99, cursor/checkpoint delay, queue and pending-ledger depth, and starvation by set. The benchmark matrix must include restart recovery, degraded quorum/scanning-disk fallback, urgent replacement heal, and at least two independent sets.
-
-The executable fixture uses a fixed eight-sample restart/degraded sequence so the baseline is reproducible without wall-clock noise: two sets each receive ScannerRead, HealRead, HealWrite and a follow-up ScannerRead. Its expected synthetic p99 is 420 microseconds, maximum modeled backlog is 2, two HealWrite samples are deferred, and the independent second set still services three reads. These are fixture values, not production SLO claims; production benchmark output must replace them with measured p99, backlog and per-set wait distributions.
-
-The inventory test reads the current source files and asserts the named guards/fallback branches are still present (`start_scan`, pending-ledger admission, Heal queue/active checks, namespace `get_write_lock`, and scanning-disk re-append). A source rename or guard removal therefore fails the baseline instead of silently leaving stale documentation.
-
-Commit-time generation-fencing, lease-expiry, and lock-order tests are intentionally deferred until a Phase-0 fixture demonstrates a stale write or an SLO violation; arithmetic-only placeholders would stay green if production paths regressed.
-
-If a future fixture demonstrates stale destructive writes, the fix must extend the storage-owned generation/admission primitive and validate the token at the final metadata/format/delete commit. Cancellation or a local lease alone is not a fence.

scripts/test/vault_ha_kms_live.sh

@@ -241,7 +241,7 @@ env \
   RUSTFS_TEST_VAULT_FAILOVER_MARKER="$MARKER" \
   RUSTFS_TEST_VAULT_OLD_LEADER="$OLD_LEADER" \
   cargo test -p rustfs-kms --test vault_ha_failover_live \
-    vault_raft_leader_failure_preserves_kv2_and_transit_decrypts -- \
+    vault_raft_leader_failure_recovers_kv2_and_transit_decrypts -- \
     --ignored --nocapture --test-threads=1 &
 TEST_PID=$!