test(heal): cover set bulkhead scheduler isolation

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

.github/workflows/docker.yml

@@ -121,6 +121,22 @@ jobs:
           create_latest=false
           source_ref="$GITHUB_SHA"
 
+          # Pre-GA policy: until the first stable (vX.Y.Z) tag exists, every
+          # prerelease (alpha/beta/rc) also moves `latest`, so users pulling
+          # `latest` get the newest test build. Once a stable tag is published
+          # this returns false and `latest` follows stable releases only.
+          prerelease_moves_latest() {
+            local stable_tags
+            stable_tags=$(git ls-remote --tags --refs origin 2>/dev/null \
+              | awk '{print $2}' \
+              | grep -E '^refs/tags/v?[0-9]+\.[0-9]+\.[0-9]+$' || true)
+            if [[ -z "$stable_tags" ]]; then
+              return 0
+            fi
+            echo "ℹ️  Stable release tag(s) already exist; prereleases no longer update latest"
+            return 1
+          }
+
           if [[ "${{ github.event_name }}" == "workflow_run" ]]; then
             # Triggered by build workflow completion
             echo "🔗 Triggered by build workflow completion"
@@ -184,8 +200,8 @@ jobs:
                 if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]] || [[ "$version" == *"rc"* ]]; then
                   build_type="prerelease"
                   is_prerelease=true
-                  # Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
-                  if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
+                  # Pre-GA policy: prereleases update latest until the first stable tag exists.
+                  if prerelease_moves_latest; then
                     create_latest=true
                     echo "🧪 Building Docker image for prerelease: $version (creating latest tag)"
                   else
@@ -243,8 +259,8 @@ jobs:
               v*alpha*|v*beta*|v*rc*|*alpha*|*beta*|*rc*)
                 build_type="prerelease"
                 is_prerelease=true
-                # Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
-                if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
+                # Pre-GA policy: prereleases update latest until the first stable tag exists.
+                if prerelease_moves_latest; then
                   create_latest=true
                   echo "🧪 Building with prerelease version: $input_version (creating latest tag)"
                 else
@@ -394,11 +410,13 @@ jobs:
           TAG_BASE="${VERSION}${VARIANT_SUFFIX}"
           TAGS="${{ env.REGISTRY_DOCKERHUB }}:$TAG_BASE,${{ env.REGISTRY_GHCR }}:$TAG_BASE,${{ env.REGISTRY_QUAY }}:$TAG_BASE"
 
-          # Add channel tags for prereleases and latest for stable
+          # Add latest when requested (stable releases, and prereleases before GA)
           if [[ "$CREATE_LATEST" == "true" ]]; then
-            # Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
             TAGS="$TAGS,${{ env.REGISTRY_DOCKERHUB }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_GHCR }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_QUAY }}:latest${VARIANT_SUFFIX}"
-          elif [[ "$BUILD_TYPE" == "prerelease" ]]; then
+          fi
+
+          # Always add the channel tag for prereleases, independent of latest
+          if [[ "$BUILD_TYPE" == "prerelease" ]]; then
             # Prerelease channel tags (alpha, beta, rc)
             if [[ "$VERSION" == *"alpha"* ]]; then
               CHANNEL="alpha"
@@ -555,7 +573,7 @@ jobs:
             "prerelease")
               echo "🧪 Prerelease Docker image has been built with ${VERSION} tags"
               echo "⚠️  This is a prerelease image - use with caution"
-              # Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
+              # Prereleases move latest until the first stable tag exists (pre-GA policy).
               if [[ "$CREATE_LATEST" == "true" ]]; then
                 echo "🏷️  Latest tag has been created for prerelease: $VERSION"
               else

Cargo.lock

@@ -6139,9 +6139,9 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
 
 [[package]]
 name = "libflate"
-version = "2.3.1"
+version = "2.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a4da9b700e758e57152a1fd1c52cbdc5727c1aa6d8743dc1acda917398f1d76c"
+checksum = "561a8da1a50e1428d3c51321dafeca849df992a5bb67720c386131234caba82e"
 dependencies = [
  "adler32",
  "crc32fast",
@@ -10943,9 +10943,9 @@ dependencies = [
 
 [[package]]
 name = "rustfs-uring"
-version = "0.2.1"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0486e62d0efe25db95c00aeacb2da84368adcba299216cda99fcb11328061c84"
+checksum = "b29bc57b4bd62a73f4fae408b536adf578332e50e464797d09dc2382c7cb68c2"
 dependencies = [
  "io-uring",
  "libc",
@@ -12406,7 +12406,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.3.4",
+ "getrandom 0.4.3",
  "once_cell",
  "rustix",
  "windows-sys 0.59.0",

crates/ecstore/Cargo.toml

@@ -226,7 +226,7 @@ metrics = { workspace = true }
 # crates.io. The guard scripts/check_no_tokio_io_uring.sh allows an explicit
 # io-uring integration; only the tokio "io-uring" runtime feature is banned.
 [target.'cfg(target_os = "linux")'.dependencies]
-rustfs-uring = "0.2.1"
+rustfs-uring = "0.2.2"
 
 [target.'cfg(windows)'.dependencies]
 winapi-util.workspace = true

crates/ecstore/src/disk/disk_store.rs

@@ -195,13 +195,6 @@ fn resolve_drive_timeout_profile_from_env() -> DriveTimeoutProfile {
     DriveTimeoutProfile::parse(rustfs_config::DEFAULT_DRIVE_TIMEOUT_PROFILE).unwrap_or(DriveTimeoutProfile::Default)
 }
 
-#[cfg(test)]
-tokio::task_local! {
-    /// Artificial `disk_info` latency for tests that pin how the admin storage
-    /// walk composes per-drive probe time.
-    pub(crate) static DISK_INFO_PROBE_DELAY_FOR_TEST: Duration;
-}
-
 fn get_drive_timeout_profile() -> DriveTimeoutProfile {
     #[cfg(test)]
     {
@@ -2043,10 +2036,6 @@ impl DiskAPI for LocalDiskWrapper {
             .track_disk_health_with_op_and_timeout_action(
                 "disk_info",
                 || async {
-                    #[cfg(test)]
-                    if let Ok(delay) = DISK_INFO_PROBE_DELAY_FOR_TEST.try_with(|delay| *delay) {
-                        tokio::time::sleep(delay).await;
-                    }
                     let result = self.disk.disk_info(opts).await?;
 
                     if let Some(current_disk_id) = *self.disk_id.read().await

crates/ecstore/src/set_disk/mod.rs

@@ -6463,114 +6463,113 @@ pub fn should_heal_object_on_disk(
     (false, false, None)
 }
 
-/// Probe every drive of the set at once. Each live probe is bounded by the
-/// drive `disk_info` timeout, and the admin peer probe budget only covers one
-/// such timeout; a sequential walk over several stalled drives after a power
-/// cut would exceed it and make healthy peers render as unknown (#6488).
 async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<rustfs_madmin::Disk> {
-    join_all(disks.iter().zip(eps).map(|(disk, ep)| disk_admin_info(disk.as_ref(), ep))).await
-}
+    let mut ret = Vec::new();
 
-async fn disk_admin_info(disk: Option<&DiskStore>, ep: &Endpoint) -> rustfs_madmin::Disk {
-    let Some(disk) = disk else {
-        return rustfs_madmin::Disk {
-            endpoint: ep.to_string(),
-            drive_path: ep.get_file_path(),
-            local: ep.is_local,
-            pool_index: ep.pool_idx,
-            set_index: ep.set_idx,
-            disk_index: ep.disk_idx,
-            runtime_state: None,
-            offline_duration_seconds: None,
-            state: DiskError::DiskNotFound.to_string(),
-            capacity_observation_source: Some("missing".to_owned()),
-            capacity_observation_age_seconds: Some(0),
-            ..Default::default()
-        };
-    };
+    for (i, pool) in disks.iter().enumerate() {
+        if let Some(disk) = pool {
+            let runtime_state = disk.runtime_state();
+            let offline_duration_seconds = disk.offline_duration_secs();
+            let capacity_snapshot = disk.last_capacity_snapshot();
+            let cached_disk_id = disk.cached_disk_id().await;
+            if runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect {
+                match disk
+                    .disk_info(&DiskInfoOptions {
+                        metrics: true,
+                        ..Default::default()
+                    })
+                    .await
+                {
+                    Ok(res) => {
+                        disk.record_capacity_probe(res.total, res.used, res.free);
+                        ret.push(rustfs_madmin::Disk {
+                            endpoint: eps[i].to_string(),
+                            local: eps[i].is_local,
+                            pool_index: eps[i].pool_idx,
+                            set_index: eps[i].set_idx,
+                            disk_index: eps[i].disk_idx,
+                            state: "ok".to_owned(),
 
-    let runtime_state = disk.runtime_state();
-    let offline_duration_seconds = disk.offline_duration_secs();
-    let capacity_snapshot = disk.last_capacity_snapshot();
-    let cached_disk_id = disk.cached_disk_id().await;
-    if !(runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect) {
-        let mut disk_info = build_runtime_snapshot_disk(ep, runtime_state, offline_duration_seconds, capacity_snapshot);
-        disk_info.metrics = disk.metrics_snapshot();
-        disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
-        return disk_info;
-    }
+                            root_disk: res.root_disk,
+                            drive_path: res.mount_path.clone(),
+                            healing: res.healing,
+                            scanning: res.scanning,
+                            runtime_state: Some(runtime_state.as_str().to_string()),
+                            offline_duration_seconds,
+                            capacity_observation_source: Some("live_probe".to_owned()),
+                            capacity_observation_age_seconds: Some(0),
 
-    match disk
-        .disk_info(&DiskInfoOptions {
-            metrics: true,
-            ..Default::default()
-        })
-        .await
-    {
-        Ok(res) => {
-            disk.record_capacity_probe(res.total, res.used, res.free);
-            rustfs_madmin::Disk {
-                endpoint: ep.to_string(),
-                local: ep.is_local,
-                pool_index: ep.pool_idx,
-                set_index: ep.set_idx,
-                disk_index: ep.disk_idx,
-                state: "ok".to_owned(),
-
-                root_disk: res.root_disk,
-                drive_path: res.mount_path.clone(),
-                healing: res.healing,
-                scanning: res.scanning,
-                runtime_state: Some(runtime_state.as_str().to_string()),
-                offline_duration_seconds,
-                capacity_observation_source: Some("live_probe".to_owned()),
-                capacity_observation_age_seconds: Some(0),
-
-                uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
-                major: res.major as u32,
-                minor: res.minor as u32,
-                model: None,
-                total_space: res.total,
-                used_space: res.used,
-                available_space: res.free,
-                physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
-                utilization: utilization_percent(res.total, res.used),
-                used_inodes: res.used_inodes,
-                free_inodes: res.free_inodes,
-                metrics: Some(res.metrics),
-                ..Default::default()
-            }
-        }
-        Err(err) => {
-            let mut disk_info = rustfs_madmin::Disk {
-                state: err.to_string(),
-                endpoint: ep.to_string(),
-                drive_path: ep.get_file_path(),
-                local: ep.is_local,
-                pool_index: ep.pool_idx,
-                set_index: ep.set_idx,
-                disk_index: ep.disk_idx,
-                runtime_state: Some(runtime_state.as_str().to_string()),
-                offline_duration_seconds,
-                metrics: disk.metrics_snapshot(),
-                uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
-                ..Default::default()
-            };
-            if let Some((total, used, free, _)) = capacity_snapshot {
-                disk_info.total_space = total;
-                disk_info.used_space = used;
-                disk_info.available_space = free;
-                disk_info.utilization = utilization_percent(total, used);
-                disk_info.capacity_observation_source = Some("snapshot".to_owned());
-                disk_info.capacity_observation_age_seconds =
-                    capacity_snapshot.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
+                            uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
+                            major: res.major as u32,
+                            minor: res.minor as u32,
+                            model: None,
+                            total_space: res.total,
+                            used_space: res.used,
+                            available_space: res.free,
+                            physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
+                            utilization: utilization_percent(res.total, res.used),
+                            used_inodes: res.used_inodes,
+                            free_inodes: res.free_inodes,
+                            metrics: Some(res.metrics),
+                            ..Default::default()
+                        });
+                    }
+                    Err(err) => {
+                        let mut disk_info = rustfs_madmin::Disk {
+                            state: err.to_string(),
+                            endpoint: eps[i].to_string(),
+                            drive_path: eps[i].get_file_path(),
+                            local: eps[i].is_local,
+                            pool_index: eps[i].pool_idx,
+                            set_index: eps[i].set_idx,
+                            disk_index: eps[i].disk_idx,
+                            runtime_state: Some(runtime_state.as_str().to_string()),
+                            offline_duration_seconds,
+                            metrics: disk.metrics_snapshot(),
+                            uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()),
+                            ..Default::default()
+                        };
+                        if let Some((total, used, free, _)) = capacity_snapshot {
+                            disk_info.total_space = total;
+                            disk_info.used_space = used;
+                            disk_info.available_space = free;
+                            disk_info.utilization = utilization_percent(total, used);
+                            disk_info.capacity_observation_source = Some("snapshot".to_owned());
+                            disk_info.capacity_observation_age_seconds = capacity_snapshot
+                                .map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
+                        } else {
+                            disk_info.capacity_observation_source = Some("missing".to_owned());
+                            disk_info.capacity_observation_age_seconds = Some(0);
+                        }
+                        ret.push(disk_info);
+                    }
+                }
             } else {
-                disk_info.capacity_observation_source = Some("missing".to_owned());
-                disk_info.capacity_observation_age_seconds = Some(0);
+                let mut disk_info =
+                    build_runtime_snapshot_disk(&eps[i], runtime_state, offline_duration_seconds, capacity_snapshot);
+                disk_info.metrics = disk.metrics_snapshot();
+                disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string());
+                ret.push(disk_info);
             }
-            disk_info
+        } else {
+            ret.push(rustfs_madmin::Disk {
+                endpoint: eps[i].to_string(),
+                drive_path: eps[i].get_file_path(),
+                local: eps[i].is_local,
+                pool_index: eps[i].pool_idx,
+                set_index: eps[i].set_idx,
+                disk_index: eps[i].disk_idx,
+                runtime_state: None,
+                offline_duration_seconds: None,
+                state: DiskError::DiskNotFound.to_string(),
+                capacity_observation_source: Some("missing".to_owned()),
+                capacity_observation_age_seconds: Some(0),
+                ..Default::default()
+            })
         }
     }
+
+    ret
 }
 
 fn build_runtime_snapshot_disk(
@@ -10692,41 +10691,6 @@ mod tests {
         );
     }
 
-    #[tokio::test(start_paused = true)]
-    async fn test_get_disks_info_probes_drives_concurrently() {
-        use crate::disk::disk_store::DISK_INFO_PROBE_DELAY_FOR_TEST;
-
-        let format = FormatV3::new(1, 4);
-        let mut temp_dirs = Vec::new();
-        let mut endpoints = Vec::new();
-        let mut disks = Vec::new();
-        for disk_idx in 0..4 {
-            let (dir, endpoint, disk) = make_formatted_local_disk_for_info_test(disk_idx, &format).await;
-            temp_dirs.push(dir);
-            endpoints.push(endpoint);
-            disks.push(Some(disk));
-        }
-
-        let probe_delay = std::time::Duration::from_secs(2);
-        let started = tokio::time::Instant::now();
-        let info = DISK_INFO_PROBE_DELAY_FOR_TEST
-            .scope(probe_delay, get_disks_info(&disks, &endpoints))
-            .await;
-        let elapsed = started.elapsed();
-
-        assert_eq!(info.len(), 4);
-        assert!(info.iter().all(|disk| disk.state == "ok"), "every drive should still report a live probe");
-        assert_eq!(
-            info.iter().map(|disk| disk.disk_index).collect::<Vec<_>>(),
-            endpoints.iter().map(|ep| ep.disk_idx).collect::<Vec<_>>(),
-            "concurrent probes must keep endpoint order"
-        );
-        assert!(
-            elapsed < probe_delay * 2,
-            "four stalled drives must cost one probe delay, not four; took {elapsed:?}"
-        );
-    }
-
     #[tokio::test]
     async fn test_get_disks_info_preserves_remote_cached_disk_id_when_offline() {
         let (endpoint, disk) = make_remote_disk_for_info_test(0).await;

crates/heal/src/heal/manager/tests.rs

@@ -944,6 +944,84 @@ fn bucket_request(bucket: &str, priority: HealPriority, source: HealRequestSourc
     request
 }
 
+fn scoped_object_request(bucket: &str, object: &str, pool_index: usize, set_index: usize) -> HealRequest {
+    HealRequest::new(
+        HealType::Object {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            version_id: None,
+        },
+        HealOptions {
+            pool_index: Some(pool_index),
+            set_index: Some(set_index),
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    )
+}
+
+#[tokio::test]
+async fn scheduler_bulkhead_starts_other_sets_and_retains_same_set_tail() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    {
+        let mut config = manager.config.write().await;
+        config.max_concurrent_heals = 2;
+        config.max_concurrent_per_set = 1;
+        config.set_bulkhead_enable = true;
+        config.event_driven_scheduler_enable = false;
+        config.mainline_throttle_enable = false;
+    }
+
+    let first_set = scoped_object_request("scheduler-bulkhead-set-a-first", "object-a", 0, 1);
+    let first_set_id = first_set.id.clone();
+    let same_set_tail = scoped_object_request("scheduler-bulkhead-set-a-tail", "object-b", 0, 1);
+    let same_set_tail_id = same_set_tail.id.clone();
+    let other_set = scoped_object_request("scheduler-bulkhead-set-b", "object-c", 0, 2);
+    let other_set_id = other_set.id.clone();
+
+    let first_hook = Arc::new(CompletedRetentionHook::default());
+    let other_hook = Arc::new(CompletedRetentionHook::default());
+    {
+        let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
+        hooks.insert("scheduler-bulkhead-set-a-first".to_string(), first_hook.clone());
+        hooks.insert("scheduler-bulkhead-set-b".to_string(), other_hook.clone());
+    }
+    {
+        let mut queue = manager.heal_queue.lock().await;
+        assert_eq!(queue.push(first_set), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(same_set_tail), QueuePushOutcome::Accepted);
+        assert_eq!(queue.push(other_set), QueuePushOutcome::Accepted);
+    }
+
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(5), first_hook.started.notified())
+        .await
+        .expect("first set task should start");
+    tokio::time::timeout(Duration::from_secs(5), other_hook.started.notified())
+        .await
+        .expect("other set task should start despite same-set tail");
+
+    assert_eq!(manager.get_active_task_count().await, 2);
+    assert_eq!(manager.get_queue_length().await, 1);
+    assert!(matches!(manager.get_task_status(&same_set_tail_id).await, Ok(HealTaskStatus::Pending)));
+    {
+        let active = manager.active_heals.lock().await;
+        assert!(active.contains_key(&first_set_id));
+        assert!(active.contains_key(&other_set_id));
+        assert!(!active.contains_key(&same_set_tail_id));
+        let counts = running_heal_set_counts(&active);
+        assert_eq!(counts.get("pool_0_set_1"), Some(&1));
+        assert_eq!(counts.get("pool_0_set_2"), Some(&1));
+    }
+
+    manager.cancel_task(&first_set_id).await.expect("cancel first active task");
+    manager.cancel_task(&other_set_id).await.expect("cancel other active task");
+    COMPLETED_RETENTION_HOOKS
+        .lock()
+        .await
+        .retain(|bucket, _| !bucket.starts_with("scheduler-bulkhead-"));
+}
+
 #[test]
 fn test_push_displacing_lower_priority_actually_enqueues_new_request() {
     // Regression for the release-build defect where the enqueue side effect lived inside