Merge branch 'cxymds/perf-decommission-checkpoint' into cxymds/perf-decommission-entry-budget

crates/config/src/constants/runtime.rs

@@ -57,13 +57,6 @@ pub const DEFAULT_MAX_IO_EVENTS_PER_TICK: usize = 1024;
 pub const DEFAULT_EVENT_INTERVAL: u32 = 61;
 pub const DEFAULT_RNG_SEED: Option<u64> = None; // None means random
 
-/// Dedicated blocking thread pool for fsync/fdatasync operations.
-/// When > 1, fsync operations are isolated from the main blocking pool to
-/// prevent device-bound fsync from starving read operations (pread/stat/open).
-/// Default 0 means auto (no isolation, use main runtime).
-pub const ENV_FSYNC_BLOCKING_THREADS: &str = "RUSTFS_RUNTIME_FSYNC_BLOCKING_THREADS";
-pub const DEFAULT_FSYNC_BLOCKING_THREADS: usize = 0;
-
 // Dial9 Tokio Telemetry Default values
 pub const DEFAULT_RUNTIME_DIAL9_ENABLED: bool = false; // Disabled by default
 pub const DEFAULT_RUNTIME_DIAL9_OUTPUT_DIR: &str = "/var/log/rustfs/telemetry";

crates/ecstore/src/disk/os.rs

@@ -315,7 +315,7 @@ pub async fn fsync_dir(dir: impl AsRef<Path>) -> io::Result<()> {
     #[cfg(unix)]
     {
         let dir = dir.as_ref().to_path_buf();
-        fsync_spawn_blocking(move || fsync_dir_std(dir)).await?
+        tokio::task::spawn_blocking(move || fsync_dir_std(dir)).await?
     }
 
     #[cfg(not(unix))]
@@ -683,7 +683,7 @@ async fn fsync_open_dst_dir_group(group: &DstDirFsyncGroup) -> io::Result<()> {
     #[cfg(test)]
     let dir = group.dir.clone();
     let dir_file = group.dir_file.clone();
-    fsync_spawn_blocking(move || {
+    tokio::task::spawn_blocking(move || {
         #[cfg(test)]
         {
             if let Some(kind) = fsync_dir_recorder::take_grouped_failure(&dir) {
@@ -1080,44 +1080,6 @@ const TEST_GLOBAL_FILE_SYNCS: usize = 64;
 
 static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
 static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
-
-/// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
-/// configured with >1 threads, isolates device-bound fsync from the main
-/// blocking pool so reads (pread/stat/open) are not starved. `None` means
-/// fall back to the main runtime (zero behavior change).
-static FSYNC_RUNTIME: LazyLock<Option<tokio::runtime::Runtime>> = LazyLock::new(|| {
-    let threads =
-        rustfs_utils::get_env_usize(rustfs_config::ENV_FSYNC_BLOCKING_THREADS, rustfs_config::DEFAULT_FSYNC_BLOCKING_THREADS);
-    if threads <= 1 {
-        return None;
-    }
-    let mut builder = tokio::runtime::Builder::new_multi_thread();
-    builder
-        .worker_threads(num_cpus::get().min(8))
-        .max_blocking_threads(threads)
-        .thread_name("rustfs-fsync")
-        .thread_stack_size(512 * 1024)
-        .enable_all();
-    match builder.build() {
-        Ok(rt) => {
-            tracing::info!(threads, "fsync dedicated blocking pool enabled");
-            Some(rt)
-        }
-        Err(err) => {
-            tracing::warn!(%err, "failed to build fsync runtime, falling back to main pool");
-            None
-        }
-    }
-});
-
-/// Spawn a blocking task on the fsync-dedicated runtime if configured,
-/// otherwise fall back to the main tokio blocking pool.
-fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'static) -> tokio::task::JoinHandle<T> {
-    match FSYNC_RUNTIME.as_ref() {
-        Some(rt) => rt.spawn_blocking(f),
-        None => tokio::task::spawn_blocking(f),
-    }
-}
 static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
     LazyLock::new(|| Mutex::new(HashMap::new()));
 type NamespaceMutationLock = AsyncMutex<()>;
@@ -1255,7 +1217,7 @@ where
     F: FnOnce() -> io::Result<T> + Send + 'static,
 {
     let (disk_permit, global_permit) = acquire_file_sync_permits(disk_permits).await?;
-    let result = fsync_spawn_blocking(move || {
+    let result = tokio::task::spawn_blocking(move || {
         let _disk_permit = disk_permit;
         work()
     })
@@ -2184,7 +2146,7 @@ async fn run_blocking_namespace_file_sync_operation_with_global<T: Send + 'stati
         wait_started,
     );
     let disk_permit = admission.disk_permit.clone();
-    let result = fsync_spawn_blocking(move || {
+    let result = tokio::task::spawn_blocking(move || {
         let _lease = lease;
         let _disk_permit = disk_permit;
         operation()

crates/ecstore/src/runtime/instance.rs

@@ -160,6 +160,10 @@ pub struct InstanceContext {
     /// workers (scanner/heal/tier/lifecycle) without touching another instance.
     /// Replaces the process-global cancel-token static.
     background_cancel_token: OnceLock<CancellationToken>,
+    /// Serializes decommission data-movement operations with cancellation and
+    /// a subsequent restart. Readers are held across one object side effect;
+    /// the transition path takes the writer after cancelling the routine.
+    decommission_operation_gate: Arc<RwLock<()>>,
     /// Resolves object-encryption material at the application boundary.
     object_encryption_resolver: OnceLock<Arc<dyn ObjectEncryptionResolver>>,
     tier_delete_journal_recovery_stores: std::sync::Mutex<HashSet<Uuid>>,
@@ -200,6 +204,7 @@ impl InstanceContext {
             local_disk_set_drives: Arc::new(RwLock::new(Vec::new())),
             bucket_metadata_sys: std::sync::Mutex::new(None),
             background_cancel_token: OnceLock::new(),
+            decommission_operation_gate: Arc::new(RwLock::new(())),
             object_encryption_resolver: OnceLock::new(),
             tier_delete_journal_recovery_stores: std::sync::Mutex::new(HashSet::new()),
             transition_transaction_recovery_stores: std::sync::Mutex::new(HashSet::new()),
@@ -218,6 +223,10 @@ impl InstanceContext {
         self.lock_manager.clone()
     }
 
+    pub(crate) fn decommission_operation_gate(&self) -> Arc<RwLock<()>> {
+        Arc::clone(&self.decommission_operation_gate)
+    }
+
     /// Install the application-owned object-encryption resolver once.
     pub fn set_object_encryption_resolver(
         &self,

rustfs/src/admin/handlers/system.rs

@@ -70,12 +70,6 @@ const SITE_REPLICATION_EDIT_ROUTE: &str = "/rustfs/admin/v3/site-replication/edi
 const SITE_REPLICATION_RESYNC_ROUTE: &str = "/rustfs/admin/v3/site-replication/resync/op";
 const SITE_REPLICATION_REPAIR_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair";
 const SITE_REPLICATION_REPAIR_STATUS_ROUTE: &str = "/rustfs/admin/v3/site-replication/repair/status";
-const IAM_POLICY_ATTACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/attach";
-const IAM_POLICY_DETACH_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy/detach";
-const IAM_POLICY_ENTITIES_ROUTE: &str = "/rustfs/admin/v3/idp/builtin/policy-entities";
-const IAM_ACCESS_KEYS_BULK_ROUTE: &str = "/rustfs/admin/v3/list-access-keys-bulk";
-const IAM_ACCESS_KEYS_BULK_LDAP_ROUTE: &str = "/rustfs/admin/v3/idp/ldap/list-access-keys-bulk";
-const IAM_ACCESS_KEYS_BULK_OPENID_ROUTE: &str = "/rustfs/admin/v3/idp/openid/list-access-keys-bulk";
 
 macro_rules! log_system_request_rejected {
     ($operation:expr, $reason:expr) => {
@@ -667,24 +661,9 @@ pub struct RuntimeCapabilitiesSummary {
     pub manual_transition_jobs: CapabilityStatus,
 }
 
-/// One named admin capability advertised to management clients
-/// (rustfs/backlog#1900). `name` is a cross-repo wire contract: the rc
-/// client gates commands on these exact strings (see rustfs/cli
-/// `IAM_POLICY_DETACH_CAPABILITY` etc.), so entries may be added but
-/// existing names must never be renamed or removed.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
-pub struct AdvertisedAdminCapability {
-    pub name: &'static str,
-    pub status: CapabilityStatus,
-}
-
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct RuntimeCapabilitiesResponse {
     pub summary: RuntimeCapabilitiesSummary,
-    /// Additive field: absent in responses from older servers, so clients
-    /// must treat a missing list as "no dynamic advertisement" and fall
-    /// back to their pinned per-version contract.
-    pub advertised: Vec<AdvertisedAdminCapability>,
     pub replication: ReplicationCapabilities,
     pub manual_transition_jobs: ManualTransitionJobCapabilities,
     pub diagnostic_probes: DiagnosticProbeCapabilities,
@@ -1007,7 +986,6 @@ pub(crate) async fn build_runtime_capabilities_response()
 
     Ok(RuntimeCapabilitiesResponse {
         summary,
-        advertised: advertised_admin_capabilities(),
         replication: ReplicationCapabilities::current(),
         manual_transition_jobs: ManualTransitionJobCapabilities::current(),
         diagnostic_probes: DiagnosticProbeCapabilities::current(),
@@ -1099,23 +1077,6 @@ fn admin_route_capability(method: HttpMethod, path: &str) -> CapabilityStatus {
     admin_route_capability_from_inventory(method, path, ADMIN_ROUTE_POLICY_SPECS, DEFERRED_ADMIN_ROUTE_POLICIES)
 }
 
-fn advertised_admin_capabilities() -> Vec<AdvertisedAdminCapability> {
-    [
-        ("admin.iam.policy-attach", HttpMethod::Post, IAM_POLICY_ATTACH_ROUTE),
-        ("admin.iam.policy-detach", HttpMethod::Post, IAM_POLICY_DETACH_ROUTE),
-        ("admin.iam.policy-entities", HttpMethod::Get, IAM_POLICY_ENTITIES_ROUTE),
-        ("admin.iam.access-keys-bulk", HttpMethod::Get, IAM_ACCESS_KEYS_BULK_ROUTE),
-        ("admin.iam.access-keys-bulk.ldap", HttpMethod::Get, IAM_ACCESS_KEYS_BULK_LDAP_ROUTE),
-        ("admin.iam.access-keys-bulk.openid", HttpMethod::Get, IAM_ACCESS_KEYS_BULK_OPENID_ROUTE),
-    ]
-    .into_iter()
-    .map(|(name, method, route)| AdvertisedAdminCapability {
-        name,
-        status: admin_route_capability(method, route),
-    })
-    .collect()
-}
-
 fn admin_route_capability_from_inventory(
     method: HttpMethod,
     path: &str,
@@ -1278,48 +1239,6 @@ mod tests {
         );
     }
 
-    /// Wire-contract pin (rustfs/backlog#1900): the rc client keys its
-    /// command gates on these exact capability names, and parses each
-    /// entry as `{name, status: {state, reason?}}`. Renaming or dropping
-    /// a name silently disables the corresponding rc command.
-    #[tokio::test]
-    async fn runtime_capabilities_response_advertises_iam_capabilities() {
-        let response = build_runtime_capabilities_response()
-            .await
-            .expect("runtime capabilities response should build");
-
-        let expected_supported = [
-            "admin.iam.policy-attach",
-            "admin.iam.policy-detach",
-            "admin.iam.policy-entities",
-            "admin.iam.access-keys-bulk",
-            "admin.iam.access-keys-bulk.ldap",
-            "admin.iam.access-keys-bulk.openid",
-        ];
-        for name in expected_supported {
-            let entry = response
-                .advertised
-                .iter()
-                .find(|capability| capability.name == name)
-                .unwrap_or_else(|| panic!("{name} must be advertised"));
-            assert_eq!(entry.status.state, CapabilityState::Supported, "{name} must be supported");
-        }
-
-        let mut names: Vec<&str> = response.advertised.iter().map(|capability| capability.name).collect();
-        let total = names.len();
-        names.sort_unstable();
-        names.dedup();
-        assert_eq!(names.len(), total, "advertised capability names must be unique");
-
-        let serialized = serde_json::to_value(&response).expect("response should serialize");
-        let advertised = serialized["advertised"].as_array().expect("advertised must be an array");
-        let detach = advertised
-            .iter()
-            .find(|entry| entry["name"] == "admin.iam.policy-detach")
-            .expect("serialized detach entry must exist");
-        assert_eq!(detach["status"]["state"], "supported");
-    }
-
     #[tokio::test]
     async fn runtime_capabilities_response_reports_missing_topology_before_storage_init() {
         let response = build_runtime_capabilities_response()